#108818 by David Strauss, chx, Crell: Add transactions to key X_save() routines.

9 files changed, 628 insertions, 399 deletions

diff --git a/includes/bootstrap.inc b/includes/bootstrap.inc
index bc13133ca..803212c14 100644
--- a/includes/bootstrap.inc
+++ b/includes/bootstrap.inc
@@ -1276,6 +1276,8 @@ function request_uri() {
  *
  * @see watchdog_severity_levels()
  * @see hook_watchdog()
+ * @see DatabaseConnection::rollback()
+ * @see DatabaseTransaction::rollback()
  */
 function watchdog($type, $message, $variables = array(), $severity = WATCHDOG_NOTICE, $link = NULL) {
   global $user, $base_root;
@@ -1602,6 +1604,10 @@ function _drupal_bootstrap_database() {
   // Initialize the database system. Note that the connection
   // won't be initialized until it is actually requested.
   require_once DRUPAL_ROOT . '/includes/database/database.inc';
+
+  // Set Drupal's watchdog as the logging callback.
+  Database::setLoggingCallback('watchdog', WATCHDOG_NOTICE, WATCHDOG_ERROR);
+
   // Register autoload functions so that we can access classes and interfaces.
   spl_autoload_register('drupal_autoload_class');
   spl_autoload_register('drupal_autoload_interface');
diff --git a/includes/database/database.inc b/includes/database/database.inc
index 81f04fb2e..074911c75 100644
--- a/includes/database/database.inc
+++ b/includes/database/database.inc
@@ -229,6 +229,13 @@ abstract class DatabaseConnection extends PDO {
   protected $willRollback;
 
   /**
+   * Array of argument arrays for logging post-rollback.
+   *
+   * @var array
+   */
+  protected $rollbackLogs = array();
+
+  /**
    * The name of the Select class for this connection.
    *
    * Normally this and the following class names would be static variables,
@@ -849,12 +856,53 @@ abstract class DatabaseConnection extends PDO {
    * Schedule the current transaction for rollback.
    *
    * This method throws an exception if no transaction is active.
-   */
-  public function rollback() {
+   *
+   * @param $type
+   *   The category to which the rollback message belongs.
+   * @param $message
+   *   The message to store in the log. Keep $message translatable
+   *   by not concatenating dynamic values into it! Variables in the
+   *   message should be added by using placeholder strings alongside
+   *   the variables argument to declare the value of the placeholders.
+   * @param $variables
+   *   Array of variables to replace in the message on display or
+   *   NULL if message is already translated or not possible to
+   *   translate.
+   * @param $severity
+   *   The severity of the message, as per RFC 3164.
+   * @param $link
+   *   A link to associate with the message.
+   *
+   * @see DatabaseTransaction::rollback()
+   * @see watchdog()
+   */
+  public function rollback($type = NULL, $message = NULL, $variables = array(), $severity = NULL, $link = NULL) {
     if ($this->transactionLayers == 0) {
       throw new NoActiveTransactionException();
     }
 
+    // Set the severity to the configured default if not specified.
+    if (!isset($severity)) {
+      $logging = Database::getLoggingCallback();
+      if (is_array($logging)) {
+        $severity = $logging['default_severity'];
+      }
+    }
+    
+    // Record in an array to send to the log after transaction rollback. Messages written
+    // directly to a log (with a database back-end) will roll back during the following
+    // transaction rollback. This is an array because rollback could be requested multiple
+    // times during a transaction, and all such errors ought to be logged.
+    if (isset($message)) {
+      $this->rollbackLogs[] = array(
+        'type' => $type,
+        'message' => $message,
+        'variables' => $variables,
+        'severity' => $severity,
+        'link' => $link,
+      );
+    }
+
     $this->willRollback = TRUE;
   }
 
@@ -890,9 +938,6 @@ abstract class DatabaseConnection extends PDO {
       if ($this->supportsTransactions()) {
         parent::beginTransaction();
       }
-
-      // Reset any scheduled rollback
-      $this->willRollback = FALSE;
     }
   }
 
@@ -912,11 +957,41 @@ abstract class DatabaseConnection extends PDO {
 
     --$this->transactionLayers;
 
-    if ($this->transactionLayers == 0 && $this->supportsTransactions()) {
+    if ($this->transactionLayers == 0) {
       if ($this->willRollback) {
-        parent::rollBack();
+        $logging = Database::getLoggingCallback();
+        $logging_callback = NULL;
+        if (is_array($logging)) {
+          $logging_callback = $logging['callback'];
+        }
+
+        if ($this->supportsTransactions()) {
+          parent::rollBack();
+        }
+        else {
+          if (isset($logging_callback)) {
+            // Log the failed rollback.
+            $logging_callback('database', 'Explicit rollback failed: not supported on active connection.', array(), $logging['error_severity']);
+          }
+          
+          // It would be nice to throw an exception here if logging failed,
+          // but throwing exceptions in destructors is not supported.
+        }
+        
+        if (isset($logging_callback)) {
+          // Play back the logged errors to the specified logging callback post-rollback.
+          foreach ($this->rollbackLogs as $log_item) {
+            $logging_callback($log_item['type'], $log_item['message'], $log_item['variables'], $log_item['severity'], $log_item['link']);
+          }
+        }
+
+        // Reset any scheduled rollback.
+        $this->willRollback = FALSE;
+  
+        // Reset the error logs.
+        $this->rollbackLogs = array();
       }
-      else {
+      elseif ($this->supportsTransactions()) {
         parent::commit();
       }
     }
@@ -1164,6 +1239,17 @@ abstract class Database {
   static protected $logs = array();
 
   /**
+   * A logging function callback array.
+   *
+   * The function must accept the same function signature as Drupal's watchdog().
+   * The array containst key/value pairs for callback (string), default_severity (int),
+   * and error_severity (int).
+   *
+   * @var string
+   */
+  static protected $logging_callback = NULL;
+
+  /**
    * Start logging a given logging key on the specified connection.
    *
    * @see DatabaseLog
@@ -1194,6 +1280,37 @@ abstract class Database {
   }
 
   /**
+   * Set a logging callback for notices and errors.
+   *
+   * @see watchdog()
+   * @param $logging_callback
+   *   The function to use as the logging callback.
+   * @param $logging_default_severity
+   *   The default severity level to use for logged messages.
+   * @param $logging_error_severity
+   *   The severity level to use for logging error messages.
+   */
+  final public static function setLoggingCallback($callback, $default_severity, $error_severity) {
+    self::$logging_callback = array(
+      'callback' => $callback,
+      'default_severity' => $default_severity,
+      'error_severity' => $error_severity,
+    );
+  }
+  
+  /**
+   * Get the logging callback for notices and errors.
+   *
+   * @return
+   *   An array with the logging callback and severity levels.
+   *
+   * @see watchdog()
+   */
+  final public static function getLoggingCallback() {
+    return self::$logging_callback;
+  }
+  
+  /**
    * Retrieve the queries logged on for given logging key.
    *
    * This method also ends logging for the specified key. To get the query log
@@ -1542,9 +1659,34 @@ class DatabaseTransaction {
    *
    * This is just a wrapper method to rollback whatever transaction stack we
    * are currently in, which is managed by the connection object itself.
-   */
-  public function rollback() {
-    $this->connection->rollback();
+   *
+   * @param $type
+   *   The category to which the rollback message belongs.
+   * @param $message
+   *   The message to store in the log. Keep $message translatable
+   *   by not concatenating dynamic values into it! Variables in the
+   *   message should be added by using placeholder strings alongside
+   *   the variables argument to declare the value of the placeholders.
+   * @param $variables
+   *   Array of variables to replace in the message on display or
+   *   NULL if message is already translated or not possible to
+   *   translate.
+   * @param $severity
+   *   The severity of the message, as per RFC 3164.
+   * @param $link
+   *   A link to associate with the message.
+   *
+   * @see DatabaseConnection::rollback()
+   * @see watchdog()
+   */
+  public function rollback($type = NULL, $message = NULL, $variables = array(), $severity = NULL, $link = NULL) {
+    if (!isset($severity)) {
+      $logging = Database::getLoggingCallback();
+      if (is_array($logging)) {
+        $severity = $logging['default_severity'];
+      }
+    }
+    $this->connection->rollback($type, $message, $variables, $severity, $link);
   }
 
   /**
diff --git a/modules/block/block.admin.inc b/modules/block/block.admin.inc
index c605424a8..a747a5b4e 100644
--- a/modules/block/block.admin.inc
+++ b/modules/block/block.admin.inc
@@ -114,6 +114,8 @@ function block_admin_display_form($form, &$form_state, $blocks, $theme) {
  * Process main blocks administration form submissions.
  */
 function block_admin_display_form_submit($form, &$form_state) {
+  $txn = db_transaction();
+
   foreach ($form_state['values'] as $block) {
     $block['status'] = (int) ($block['region'] != BLOCK_REGION_NONE);
     $block['region'] = $block['status'] ? $block['region'] : '';
@@ -365,6 +367,8 @@ function block_admin_configure_validate($form, &$form_state) {
 
 function block_admin_configure_submit($form, &$form_state) {
   if (!form_get_errors()) {
+    $txn = db_transaction();
+
     db_update('block')
       ->fields(array(
         'visibility' => (int) $form_state['values']['visibility'],
diff --git a/modules/comment/comment.module b/modules/comment/comment.module
index 0d9f9c7ed..5b10d35be 100644
--- a/modules/comment/comment.module
+++ b/modules/comment/comment.module
@@ -1245,144 +1245,151 @@ function comment_access($op, $comment) {
 function comment_save($comment) {
   global $user;
 
-  $defaults =  array(
-    'mail' => '',
-    'homepage' => '',
-    'name' => '',
-    'status' => user_access('post comments without approval') ? COMMENT_PUBLISHED : COMMENT_NOT_PUBLISHED,
-  );
-  foreach ($defaults as $key => $default) {
-    if (!isset($comment->$key)) {
-      $comment->$key = $default;
+  $transaction = db_transaction();
+  try {
+    $defaults =  array(
+      'mail' => '',
+      'homepage' => '',
+      'name' => '',
+      'status' => user_access('post comments without approval') ? COMMENT_PUBLISHED : COMMENT_NOT_PUBLISHED,
+    );
+    foreach ($defaults as $key => $default) {
+      if (!isset($comment->$key)) {
+        $comment->$key = $default;
+      }
+    }
+    // Make sure we have a bundle name.
+    if (!isset($comment->node_type)) {
+      $node = node_load($comment->nid);
+      $comment->node_type = 'comment_node_' . $node->type;
     }
-  }
-  // Make sure we have a bundle name.
-  if (!isset($comment->node_type)) {
-    $node = node_load($comment->nid);
-    $comment->node_type = 'comment_node_' . $node->type;
-  }
 
-  field_attach_presave('comment', $comment);
+    field_attach_presave('comment', $comment);
 
-  // Allow modules to alter the comment before saving.
-  module_invoke_all('comment_presave', $comment);
+    // Allow modules to alter the comment before saving.
+    module_invoke_all('comment_presave', $comment);
 
-  if ($comment->cid) {
-    // Update the comment in the database.
-    db_update('comment')
-      ->fields(array(
-        'status' => $comment->status,
-        'created' => $comment->created,
-        'changed' => $comment->changed,
-        'subject' => $comment->subject,
-        'comment' => $comment->comment,
-        'format' => $comment->comment_format,
-        'uid' => $comment->uid,
-        'name' => $comment->name,
-        'mail' => $comment->mail,
-        'homepage' => $comment->homepage,
-        'language' => $comment->language,
-      ))
-      ->condition('cid', $comment->cid)
-      ->execute();
-    field_attach_update('comment', $comment);
-    // Allow modules to respond to the updating of a comment.
-    module_invoke_all('comment_update', $comment);
-    // Add an entry to the watchdog log.
-    watchdog('content', 'Comment: updated %subject.', array('%subject' => $comment->subject), WATCHDOG_NOTICE, l(t('view'), 'comment/' . $comment->cid, array('fragment' => 'comment-' . $comment->cid)));
-  }
-  else {
-    // Add the comment to database. This next section builds the thread field.
-    // Also see the documentation for comment_build().
-    if ($comment->pid == 0) {
-      // This is a comment with no parent comment (depth 0): we start
-      // by retrieving the maximum thread level.
-      $max = db_query('SELECT MAX(thread) FROM {comment} WHERE nid = :nid', array(':nid' => $comment->nid))->fetchField();
-      // Strip the "/" from the end of the thread.
-      $max = rtrim($max, '/');
-      // Finally, build the thread field for this new comment.
-      $thread = int2vancode(vancode2int($max) + 1) . '/';
+    if ($comment->cid) {
+      // Update the comment in the database.
+      db_update('comment')
+        ->fields(array(
+          'status' => $comment->status,
+          'created' => $comment->created,
+          'changed' => $comment->changed,
+          'subject' => $comment->subject,
+          'comment' => $comment->comment,
+          'format' => $comment->comment_format,
+          'uid' => $comment->uid,
+          'name' => $comment->name,
+          'mail' => $comment->mail,
+          'homepage' => $comment->homepage,
+          'language' => $comment->language,
+        ))
+        ->condition('cid', $comment->cid)
+        ->execute();
+      field_attach_update('comment', $comment);
+      // Allow modules to respond to the updating of a comment.
+      module_invoke_all('comment_update', $comment);
+      // Add an entry to the watchdog log.
+      watchdog('content', 'Comment: updated %subject.', array('%subject' => $comment->subject), WATCHDOG_NOTICE, l(t('view'), 'comment/' . $comment->cid, array('fragment' => 'comment-' . $comment->cid)));
     }
     else {
-      // This is a comment with a parent comment, so increase the part of the
-      // thread value at the proper depth.
-
-      // Get the parent comment:
-      $parent = comment_load($comment->pid);
-      // Strip the "/" from the end of the parent thread.
-      $parent->thread = (string) rtrim((string) $parent->thread, '/');
-      // Get the max value in *this* thread.
-      $max = db_query("SELECT MAX(thread) FROM {comment} WHERE thread LIKE :thread AND nid = :nid", array(
-        ':thread' => $parent->thread . '.%',
-        ':nid' => $comment->nid,
-      ))->fetchField();
-
-      if ($max == '') {
-        // First child of this parent.
-        $thread = $parent->thread . '.' . int2vancode(0) . '/';
-      }
-      else {
-        // Strip the "/" at the end of the thread.
+      // Add the comment to database. This next section builds the thread field.
+      // Also see the documentation for comment_build().
+      if ($comment->pid == 0) {
+        // This is a comment with no parent comment (depth 0): we start
+        // by retrieving the maximum thread level.
+        $max = db_query('SELECT MAX(thread) FROM {comment} WHERE nid = :nid', array(':nid' => $comment->nid))->fetchField();
+        // Strip the "/" from the end of the thread.
         $max = rtrim($max, '/');
-        // Get the value at the correct depth.
-        $parts = explode('.', $max);
-        $parent_depth = count(explode('.', $parent->thread));
-        $last = $parts[$parent_depth];
         // Finally, build the thread field for this new comment.
-        $thread = $parent->thread . '.' . int2vancode(vancode2int($last) + 1) . '/';
+        $thread = int2vancode(vancode2int($max) + 1) . '/';
+      }
+      else {
+        // This is a comment with a parent comment, so increase the part of the
+        // thread value at the proper depth.
+
+        // Get the parent comment:
+        $parent = comment_load($comment->pid);
+        // Strip the "/" from the end of the parent thread.
+        $parent->thread = (string) rtrim((string) $parent->thread, '/');
+        // Get the max value in *this* thread.
+        $max = db_query("SELECT MAX(thread) FROM {comment} WHERE thread LIKE :thread AND nid = :nid", array(
+          ':thread' => $parent->thread . '.%',
+          ':nid' => $comment->nid,
+        ))->fetchField();
+
+        if ($max == '') {
+          // First child of this parent.
+          $thread = $parent->thread . '.' . int2vancode(0) . '/';
+        }
+        else {
+          // Strip the "/" at the end of the thread.
+          $max = rtrim($max, '/');
+          // Get the value at the correct depth.
+          $parts = explode('.', $max);
+          $parent_depth = count(explode('.', $parent->thread));
+          $last = $parts[$parent_depth];
+          // Finally, build the thread field for this new comment.
+          $thread = $parent->thread . '.' . int2vancode(vancode2int($last) + 1) . '/';
+        }
       }
-    }
 
-    if (empty($comment->created)) {
-      $comment->created = REQUEST_TIME;
-    }
+      if (empty($comment->created)) {
+        $comment->created = REQUEST_TIME;
+      }
 
-    if (empty($comment->changed)) {
-      $comment->changed = $comment->created;
-    }
+      if (empty($comment->changed)) {
+        $comment->changed = $comment->created;
+      }
 
-    if ($comment->uid === $user->uid && isset($user->name)) { // '===' Need to modify anonymous users as well.
-      $comment->name = $user->name;
-    }
+      if ($comment->uid === $user->uid && isset($user->name)) { // '===' Need to modify anonymous users as well.
+        $comment->name = $user->name;
+      }
 
-    $comment->cid = db_insert('comment')
-      ->fields(array(
-        'nid' => $comment->nid,
-        'pid' => empty($comment->pid) ? 0 : $comment->pid,
-        'uid' => $comment->uid,
-        'subject' => $comment->subject,
-        'comment' => $comment->comment,
-        'format' => $comment->comment_format,
-        'hostname' => ip_address(),
-        'created' => $comment->created,
-        'changed' => $comment->changed,
-        'status' => $comment->status,
-        'thread' => $thread,
-        'name' => $comment->name,
-        'mail' => $comment->mail,
-        'homepage' => $comment->homepage,
-        'language' => $comment->language,
-      ))
-      ->execute();
+      $comment->cid = db_insert('comment')
+        ->fields(array(
+          'nid' => $comment->nid,
+          'pid' => empty($comment->pid) ? 0 : $comment->pid,
+          'uid' => $comment->uid,
+          'subject' => $comment->subject,
+          'comment' => $comment->comment,
+          'format' => $comment->comment_format,
+          'hostname' => ip_address(),
+          'created' => $comment->created,
+          'changed' => $comment->changed,
+          'status' => $comment->status,
+          'thread' => $thread,
+          'name' => $comment->name,
+          'mail' => $comment->mail,
+          'homepage' => $comment->homepage,
+          'language' => $comment->language,
+        ))
+        ->execute();
 
-    // Ignore slave server temporarily to give time for the
-    // saved node to be propagated to the slave.
-    db_ignore_slave();
+      // Ignore slave server temporarily to give time for the
+      // saved node to be propagated to the slave.
+      db_ignore_slave();
 
-    field_attach_insert('comment', $comment);
+      field_attach_insert('comment', $comment);
 
-    // Tell the other modules a new comment has been submitted.
-    module_invoke_all('comment_insert', $comment);
-    // Add an entry to the watchdog log.
-    watchdog('content', 'Comment: added %subject.', array('%subject' => $comment->subject), WATCHDOG_NOTICE, l(t('view'), 'comment/' . $comment->cid, array('fragment' => 'comment-' . $comment->cid)));
-  }
-  _comment_update_node_statistics($comment->nid);
-  // Clear the cache so an anonymous user can see his comment being added.
-  cache_clear_all();
+      // Tell the other modules a new comment has been submitted.
+      module_invoke_all('comment_insert', $comment);
+      // Add an entry to the watchdog log.
+      watchdog('content', 'Comment: added %subject.', array('%subject' => $comment->subject), WATCHDOG_NOTICE, l(t('view'), 'comment/' . $comment->cid, array('fragment' => 'comment-' . $comment->cid)));
+    }
+    _comment_update_node_statistics($comment->nid);
+    // Clear the cache so an anonymous user can see his comment being added.
+    cache_clear_all();
 
-  if ($comment->status == COMMENT_PUBLISHED) {
-    module_invoke_all('comment_publish', $comment);
+    if ($comment->status == COMMENT_PUBLISHED) {
+      module_invoke_all('comment_publish', $comment);
+    }
+  }
+  catch (Exception $e) {
+    $transaction->rollback('comment', $e->getMessage(), array(), WATCHDOG_ERROR);
   }
+  
 }
 
 /**
diff --git a/modules/node/node.module b/modules/node/node.module
index 4b036225b..e3775ca64 100644
--- a/modules/node/node.module
+++ b/modules/node/node.module
@@ -933,108 +933,115 @@ function node_submit($node) {
  *   omitted (or $node->is_new is TRUE), a new node will be added.
  */
 function node_save($node) {
-  field_attach_presave('node', $node);
-  // Let modules modify the node before it is saved to the database.
-  module_invoke_all('node_presave', $node);
-  global $user;
+  $transaction = db_transaction();
 
-  if (!isset($node->is_new)) {
-    $node->is_new = empty($node->nid);
-  }
+  try {
+    field_attach_presave('node', $node);
+    // Let modules modify the node before it is saved to the database.
+    module_invoke_all('node_presave', $node);
+    global $user;
 
-  // Apply filters to some default node fields:
-  if ($node->is_new) {
-    // Insert a new node.
-    $node->is_new = TRUE;
+    if (!isset($node->is_new)) {
+      $node->is_new = empty($node->nid);
+    }
 
-    // When inserting a node, $node->log must be set because
-    // {node_revision}.log does not (and cannot) have a default
-    // value. If the user does not have permission to create
-    // revisions, however, the form will not contain an element for
-    // log so $node->log will be unset at this point.
-    if (!isset($node->log)) {
-      $node->log = '';
+    // Apply filters to some default node fields:
+    if ($node->is_new) {
+      // Insert a new node.
+      $node->is_new = TRUE;
+
+      // When inserting a node, $node->log must be set because
+      // {node_revision}.log does not (and cannot) have a default
+      // value. If the user does not have permission to create
+      // revisions, however, the form will not contain an element for
+      // log so $node->log will be unset at this point.
+      if (!isset($node->log)) {
+        $node->log = '';
+      }
     }
-  }
-  elseif (!empty($node->revision)) {
-    $node->old_vid = $node->vid;
-    unset($node->vid);
-  }
-  else {
-    // When updating a node, avoid clobbering an existing log entry with an empty one.
-    if (empty($node->log)) {
-      unset($node->log);
+    elseif (!empty($node->revision)) {
+      $node->old_vid = $node->vid;
+      unset($node->vid);
+    }
+    else {
+      // When updating a node, avoid clobbering an existing log entry with an empty one.
+      if (empty($node->log)) {
+        unset($node->log);
+      }
     }
-  }
-
-  // Set some required fields:
-  if (empty($node->created)) {
-    $node->created = REQUEST_TIME;
-  }
-  // The changed timestamp is always updated for bookkeeping purposes (revisions, searching, ...)
-  $node->changed = REQUEST_TIME;
-
-  $node->timestamp = REQUEST_TIME;
-  $update_node = TRUE;
-
-  // When converting the title property to fields we preserved the {node}.title
-  // db column for performance, setting the default language value into this
-  // column. After this we restore the field data structure to the previous node
-  // title field.
-  $title_field = $node->title;
-  $langcode = FIELD_LANGUAGE_NONE;
-  $node->title = $title_field[$langcode][0]['value'];
 
-  // Generate the node table query and the node_revisions table query.
-  if ($node->is_new) {
-    drupal_write_record('node', $node);
-    _node_save_revision($node, $user->uid);
-    $op = 'insert';
-  }
-  else {
-    drupal_write_record('node', $node, 'nid');
-    if (!empty($node->revision)) {
+    // Set some required fields:
+    if (empty($node->created)) {
+      $node->created = REQUEST_TIME;
+    }
+    // The changed timestamp is always updated for bookkeeping purposes (revisions, searching, ...)
+    $node->changed = REQUEST_TIME;
+  
+    $node->timestamp = REQUEST_TIME;
+    $update_node = TRUE;
+
+    // When converting the title property to fields we preserved the {node}.title
+    // db column for performance, setting the default language value into this
+    // column. After this we restore the field data structure to the previous node
+    // title field.
+    $title_field = $node->title;
+    $langcode = FIELD_LANGUAGE_NONE;
+    $node->title = $title_field[$langcode][0]['value'];
+
+    // Generate the node table query and the node_revisions table query.
+    if ($node->is_new) {
+      drupal_write_record('node', $node);
       _node_save_revision($node, $user->uid);
+      $op = 'insert';
     }
     else {
-      _node_save_revision($node, $user->uid, 'vid');
-      $update_node = FALSE;
+      drupal_write_record('node', $node, 'nid');
+      if (!empty($node->revision)) {
+        _node_save_revision($node, $user->uid);
+      }
+      else {
+        _node_save_revision($node, $user->uid, 'vid');
+        $update_node = FALSE;
+      }
+      $op = 'update';
     }
-    $op = 'update';
+    if ($update_node) {
+      db_update('node')
+        ->fields(array('vid' => $node->vid))
+        ->condition('nid', $node->nid)
+        ->execute();
+    }
+  
+    // Restore the title field data structure after db storage.
+    $node->title = $title_field;
+  
+    // Call the node specific callback (if any). This can be
+    // node_invoke($node, 'insert') or
+    // node_invoke($node, 'update').
+    node_invoke($node, $op);
+  
+    // Save fields.
+    $function = "field_attach_$op";
+    $function('node', $node);
+  
+    module_invoke_all('node_' . $op, $node);
+  
+    // Update the node access table for this node.
+    node_access_acquire_grants($node);
+  
+    // Clear internal properties.
+    unset($node->is_new);
+  
+    // Clear the page and block caches.
+    cache_clear_all();
+  
+    // Ignore slave server temporarily to give time for the
+    // saved node to be propagated to the slave.
+    db_ignore_slave();
   }
-  if ($update_node) {
-    db_update('node')
-      ->fields(array('vid' => $node->vid))
-      ->condition('nid', $node->nid)
-      ->execute();
+  catch (Exception $e) {
+    $transaction->rollback('node', $e->getMessage(), array(), WATCHDOG_ERROR);
   }
-
-  // Restore the title field data structure after db storage.
-  $node->title = $title_field;
-
-  // Call the node specific callback (if any). This can be
-  // node_invoke($node, 'insert') or
-  // node_invoke($node, 'update').
-  node_invoke($node, $op);
-
-  // Save fields.
-  $function = "field_attach_$op";
-  $function('node', $node);
-
-  module_invoke_all('node_' . $op, $node);
-
-  // Update the node access table for this node.
-  node_access_acquire_grants($node);
-
-  // Clear internal properties.
-  unset($node->is_new);
-
-  // Clear the page and block caches.
-  cache_clear_all();
-
-  // Ignore slave server temporarily to give time for the
-  // saved node to be propagated to the slave.
-  db_ignore_slave();
 }
 
 /**
diff --git a/modules/node/node.test b/modules/node/node.test
index 807dc5f65..2fd36a97c 100644
--- a/modules/node/node.test
+++ b/modules/node/node.test
@@ -329,7 +329,8 @@ class PageCreationTestCase extends DrupalWebTestCase {
   }
 
   function setUp() {
-    parent::setUp();
+    // Enable dummy module that implements hook_node_post_save for exceptions.
+    parent::setUp('node_test_exception');
 
     $web_user = $this->drupalCreateUser(array('create page content', 'edit own page content'));
     $this->drupalLogin($web_user);
@@ -353,6 +354,37 @@ class PageCreationTestCase extends DrupalWebTestCase {
     $node = $this->drupalGetNodeByTitle($edit["title[$langcode][0][value]"]);
     $this->assertTrue($node, t('Node found in database.'));
   }
+
+  /**
+   * Create a page node and verify that a transaction rolls back the failed creation
+   */
+  function testFailedPageCreation() {
+    // Create a node.
+    $edit = array();
+    $langcode = FIELD_LANGUAGE_NONE;
+    $edit["title[$langcode][0][value]"] = 'testing_transaction_exception';
+    $edit["body[$langcode][0][value]"] = $this->randomName(16);
+    $this->drupalPost('node/add/page', $edit, t('Save'));
+
+    if (Database::getConnection()->supportsTransactions()) {
+      // Check that the node does not exist in the database.
+      $node = $this->drupalGetNodeByTitle($edit["title[$langcode][0][value]"]);
+      $this->assertFalse($node, t('Transactions supported, and node not found in database.'));
+    }
+    else {
+      // Check that the node exists in the database.
+      $node = $this->drupalGetNodeByTitle($edit["title[$langcode][0][value]"]);
+      $this->assertTrue($node, t('Transactions not supported, and node found in database.'));
+
+      // Check that the failed rollback was logged.
+      $records = db_query("SELECT wid FROM {watchdog} WHERE message LIKE 'Explicit rollback failed%'")->fetchAll();
+      $this->assertTrue(count($records) > 0, t('Transactions not supported, and rollback error logged to watchdog.'));      
+    }
+
+    // Check that the rollback error was logged.
+    $records = db_query("SELECT wid FROM {watchdog} WHERE message LIKE 'Test exception for rollback.'")->fetchAll();
+    $this->assertTrue(count($records) > 0, t('Rollback explanatory error logged to watchdog.'));
+  }
 }
 
 class PageViewTestCase extends DrupalWebTestCase {
diff --git a/modules/node/tests/node_test_exception.info b/modules/node/tests/node_test_exception.info
new file mode 100644
index 000000000..afe5b3719
--- /dev/null
+++ b/modules/node/tests/node_test_exception.info
@@ -0,0 +1,8 @@
+; $Id$
+name = "Node module exception tests"
+description = "Support module for node related exception testing."
+package = Testing
+version = VERSION
+core = 7.x
+files[] = node_test_exception.module
+hidden = TRUE
diff --git a/modules/node/tests/node_test_exception.module b/modules/node/tests/node_test_exception.module
new file mode 100644
index 000000000..7b09c77c7
--- /dev/null
+++ b/modules/node/tests/node_test_exception.module
@@ -0,0 +1,17 @@
+<?php
+// $Id$
+
+/**
+ * @file
+ * Dummy module implementing node related hooks to test API interaction with
+ * the Node module.
+ */
+
+/**
+ * Implement hook_node_insert().
+ */
+function node_test_exception_node_insert($node) {
+  if ($node->title['zxx'][0]['value'] == 'testing_transaction_exception') {
+    throw new Exception('Test exception for rollback.');
+  }
+}
diff --git a/modules/user/user.module b/modules/user/user.module
index a9cca9849..2772b5ffb 100644
--- a/modules/user/user.module
+++ b/modules/user/user.module
@@ -304,219 +304,225 @@ function user_load_by_name($name) {
  *   A fully-loaded $user object upon successful save or FALSE if the save failed.
  */
 function user_save($account, $edit = array(), $category = 'account') {
-  $table = drupal_get_schema('users');
-  $user_fields = $table['fields'];
-
-  if (!empty($edit['pass'])) {
-    // Allow alternate password hashing schemes.
-    require_once DRUPAL_ROOT . '/' . variable_get('password_inc', 'includes/password.inc');
-    $edit['pass'] = user_hash_password(trim($edit['pass']));
-    // Abort if the hashing failed and returned FALSE.
-    if (!$edit['pass']) {
-      return FALSE;
+  $transaction = db_transaction();
+  try {
+    $table = drupal_get_schema('users');
+    $user_fields = $table['fields'];
+
+    if (!empty($edit['pass'])) {
+      // Allow alternate password hashing schemes.
+      require_once DRUPAL_ROOT . '/' . variable_get('password_inc', 'includes/password.inc');
+      $edit['pass'] = user_hash_password(trim($edit['pass']));
+      // Abort if the hashing failed and returned FALSE.
+      if (!$edit['pass']) {
+        return FALSE;
+      }
+    }
+    else {
+      // Avoid overwriting an existing password with a blank password.
+      unset($edit['pass']);
     }
-  }
-  else {
-    // Avoid overwriting an existing password with a blank password.
-    unset($edit['pass']);
-  }
 
-  // Get the fields form so we can recognize the fields in the $edit
-  // form that should not go into the serialized data array.
-  $field_form = array();
-  $field_form_state = array();
-  $edit = (object) $edit;
-  field_attach_form('user', $edit, $field_form, $field_form_state);
+    // Get the fields form so we can recognize the fields in the $edit
+    // form that should not go into the serialized data array.
+    $field_form = array();
+    $field_form_state = array();
+    $edit = (object) $edit;
+    field_attach_form('user', $edit, $field_form, $field_form_state);
 
-  // Presave fields.
-  field_attach_presave('user', $edit);
+    // Presave fields.
+    field_attach_presave('user', $edit);
 
-  $edit = (array) $edit;
+    $edit = (array) $edit;
 
-  if (!isset($account->is_new)) {
-    $account->is_new = empty($account->uid);
-  }
+    if (!isset($account->is_new)) {
+      $account->is_new = empty($account->uid);
+    }
 
-  user_module_invoke('presave', $edit, $account, $category);
+    user_module_invoke('presave', $edit, $account, $category);
 
-  if (is_object($account) && !$account->is_new) {
-    $data = unserialize(db_query('SELECT data FROM {users} WHERE uid = :uid', array(':uid' => $account->uid))->fetchField());
-    // Consider users edited by an administrator as logged in, if they haven't
-    // already, so anonymous users can view the profile (if allowed).
-    if (empty($edit['access']) && empty($account->access) && user_access('administer users')) {
-      $edit['access'] = REQUEST_TIME;
-    }
-    foreach ($edit as $key => $value) {
-      // Form fields that don't pertain to the users, user_roles, or
-      // Field API are automatically serialized into the users.data
-      // column.
-      if (!in_array($key, array('roles', 'is_new')) && empty($user_fields[$key]) && empty($field_form[$key])) {
-        if ($value === NULL) {
-          unset($data[$key]);
-        }
-        else {
-          $data[$key] = $value;
+    if (is_object($account) && !$account->is_new) {
+      $data = unserialize(db_query('SELECT data FROM {users} WHERE uid = :uid', array(':uid' => $account->uid))->fetchField());
+      // Consider users edited by an administrator as logged in, if they haven't
+      // already, so anonymous users can view the profile (if allowed).
+      if (empty($edit['access']) && empty($account->access) && user_access('administer users')) {
+        $edit['access'] = REQUEST_TIME;
+      }
+      foreach ($edit as $key => $value) {
+        // Form fields that don't pertain to the users, user_roles, or
+        // Field API are automatically serialized into the users.data
+        // column.
+        if (!in_array($key, array('roles', 'is_new')) && empty($user_fields[$key]) && empty($field_form[$key])) {
+          if ($value === NULL) {
+            unset($data[$key]);
+          }
+          else {
+            $data[$key] = $value;
+          }
         }
       }
-    }
 
-    // Process picture uploads.
-    if (!empty($edit['picture']->fid)) {
-      $picture = $edit['picture'];
-      // If the picture is a temporary file move it to its final location and
-      // make it permanent.
-      if (($picture->status & FILE_STATUS_PERMANENT) == 0) {
-        $info = image_get_info($picture->uri);
-        $picture_directory =  variable_get('file_default_scheme', 'public') . '://' . variable_get('user_picture_path', 'pictures');
-
-        // Prepare the pictures directory.
-        file_prepare_directory($picture_directory, FILE_CREATE_DIRECTORY);
-        $destination = file_stream_wrapper_uri_normalize($picture_directory . '/picture-' . $account->uid . '.' . $info['extension']);
-
-        if ($picture = file_move($picture, $destination, FILE_EXISTS_REPLACE)) {
-          $picture->status |= FILE_STATUS_PERMANENT;
-          $edit['picture'] = file_save($picture);
+      // Process picture uploads.
+      if (!empty($edit['picture']->fid)) {
+        $picture = $edit['picture'];
+        // If the picture is a temporary file move it to its final location and
+        // make it permanent.
+        if (($picture->status & FILE_STATUS_PERMANENT) == 0) {
+          $info = image_get_info($picture->uri);
+          $picture_directory =  variable_get('file_default_scheme', 'public') . '://' . variable_get('user_picture_path', 'pictures');
+
+          // Prepare the pictures directory.
+          file_prepare_directory($picture_directory, FILE_CREATE_DIRECTORY);
+          $destination = file_stream_wrapper_uri_normalize($picture_directory . '/picture-' . $account->uid . '.' . $info['extension']);
+
+          if ($picture = file_move($picture, $destination, FILE_EXISTS_REPLACE)) {
+            $picture->status |= FILE_STATUS_PERMANENT;
+            $edit['picture'] = file_save($picture);
+          }
         }
       }
-    }
-    $edit['picture'] = empty($edit['picture']->fid) ? 0 : $edit['picture']->fid;
-
-    $edit['data'] = $data;
-    // Do not allow 'uid' to be changed.
-    $edit['uid'] = $account->uid;
-    // Save changes to the user table.
-    $success = drupal_write_record('users', $edit, 'uid');
-    if ($success === FALSE) {
-      // The query failed - better to abort the save than risk further
-      // data loss.
-      return FALSE;
-    }
-
-    // If the picture changed or was unset, remove the old one. This step needs
-    // to occur after updating the {users} record so that user_file_references()
-    // doesn't report it in use and block the deletion.
-    if (!empty($account->picture->fid) && ($edit['picture'] != $account->picture->fid)) {
-      file_delete($account->picture);
-    }
+      $edit['picture'] = empty($edit['picture']->fid) ? 0 : $edit['picture']->fid;
+
+      $edit['data'] = $data;
+      // Do not allow 'uid' to be changed.
+      $edit['uid'] = $account->uid;
+      // Save changes to the user table.
+      $success = drupal_write_record('users', $edit, 'uid');
+      if ($success === FALSE) {
+        // The query failed - better to abort the save than risk further
+        // data loss.
+        return FALSE;
+      }
 
-    // Reload user roles if provided.
-    if (isset($edit['roles']) && is_array($edit['roles'])) {
-      db_delete('users_roles')
-        ->condition('uid', $account->uid)
-        ->execute();
+      // If the picture changed or was unset, remove the old one. This step needs
+      // to occur after updating the {users} record so that user_file_references()
+      // doesn't report it in use and block the deletion.
+      if (!empty($account->picture->fid) && ($edit['picture'] != $account->picture->fid)) {
+        file_delete($account->picture);
+      }
 
-      $query = db_insert('users_roles')->fields(array('uid', 'rid'));
-      foreach (array_keys($edit['roles']) as $rid) {
-        if (!in_array($rid, array(DRUPAL_ANONYMOUS_RID, DRUPAL_AUTHENTICATED_RID))) {
-          $query->values(array(
-            'uid' => $account->uid,
-            'rid' => $rid,
-          ));
+      // Reload user roles if provided.
+      if (isset($edit['roles']) && is_array($edit['roles'])) {
+        db_delete('users_roles')
+          ->condition('uid', $account->uid)
+          ->execute();
+
+        $query = db_insert('users_roles')->fields(array('uid', 'rid'));
+        foreach (array_keys($edit['roles']) as $rid) {
+          if (!in_array($rid, array(DRUPAL_ANONYMOUS_RID, DRUPAL_AUTHENTICATED_RID))) {
+            $query->values(array(
+              'uid' => $account->uid,
+              'rid' => $rid,
+            ));
+          }
         }
+        $query->execute();
       }
-      $query->execute();
-    }
 
-    // Delete a blocked user's sessions to kick them if they are online.
-    if (isset($edit['status']) && $edit['status'] == 0) {
-      drupal_session_destroy_uid($account->uid);
-    }
+      // Delete a blocked user's sessions to kick them if they are online.
+      if (isset($edit['status']) && $edit['status'] == 0) {
+        drupal_session_destroy_uid($account->uid);
+      }
 
-    // If the password changed, delete all open sessions and recreate
-    // the current one.
-    if (!empty($edit['pass'])) {
-      drupal_session_destroy_uid($account->uid);
-      if ($account->uid == $GLOBALS['user']->uid) {
-        drupal_session_regenerate();
+      // If the password changed, delete all open sessions and recreate
+      // the current one.
+      if (!empty($edit['pass'])) {
+        drupal_session_destroy_uid($account->uid);
+        if ($account->uid == $GLOBALS['user']->uid) {
+          drupal_session_regenerate();
+        }
       }
-    }
 
-    // Save Field data.
-    $object = (object) $edit;
-    field_attach_update('user', $object);
+      // Save Field data.
+      $object = (object) $edit;
+      field_attach_update('user', $object);
 
-    // Refresh user object.
-    $user = user_load($account->uid, TRUE);
+      // Refresh user object.
+      $user = user_load($account->uid, TRUE);
 
-    // Send emails after we have the new user object.
-    if (isset($edit['status']) && $edit['status'] != $account->status) {
-      // The user's status is changing; conditionally send notification email.
-      $op = $edit['status'] == 1 ? 'status_activated' : 'status_blocked';
-      _user_mail_notify($op, $user);
-    }
+      // Send emails after we have the new user object.
+      if (isset($edit['status']) && $edit['status'] != $account->status) {
+        // The user's status is changing; conditionally send notification email.
+        $op = $edit['status'] == 1 ? 'status_activated' : 'status_blocked';
+        _user_mail_notify($op, $user);
+      }
 
-    user_module_invoke('update', $edit, $user, $category);
-  }
-  else {
-    // Allow 'uid' to be set by the caller. There is no danger of writing an
-    // existing user as drupal_write_record will do an INSERT.
-    if (empty($edit['uid'])) {
-      $edit['uid'] = db_next_id(db_query('SELECT MAX(uid) FROM {users}')->fetchField());
-    }
-    // Allow 'created' to be set by the caller.
-    if (!isset($edit['created'])) {
-      $edit['created'] = REQUEST_TIME;
-    }
-    // Consider users created by an administrator as already logged in, so
-    // anonymous users can view the profile (if allowed).
-    if (empty($edit['access']) && user_access('administer users')) {
-      $edit['access'] = REQUEST_TIME;
+      user_module_invoke('update', $edit, $user, $category);
     }
+    else {
+      // Allow 'uid' to be set by the caller. There is no danger of writing an
+      // existing user as drupal_write_record will do an INSERT.
+      if (empty($edit['uid'])) {
+        $edit['uid'] = db_next_id(db_query('SELECT MAX(uid) FROM {users}')->fetchField());
+      }
+      // Allow 'created' to be set by the caller.
+      if (!isset($edit['created'])) {
+        $edit['created'] = REQUEST_TIME;
+      }
+      // Consider users created by an administrator as already logged in, so
+      // anonymous users can view the profile (if allowed).
+      if (empty($edit['access']) && user_access('administer users')) {
+        $edit['access'] = REQUEST_TIME;
+      }
 
-    $edit['mail'] = trim($edit['mail']);
-    $success = drupal_write_record('users', $edit);
-    if ($success === FALSE) {
-      // On a failed INSERT some other existing user's uid may be returned.
-      // We must abort to avoid overwriting their account.
-      return FALSE;
-    }
+      $edit['mail'] = trim($edit['mail']);
+      $success = drupal_write_record('users', $edit);
+      if ($success === FALSE) {
+        // On a failed INSERT some other existing user's uid may be returned.
+        // We must abort to avoid overwriting their account.
+        return FALSE;
+      }
 
-    // Build the initial user object.
-    $user = user_load($edit['uid'], TRUE);
+      // Build the initial user object.
+      $user = user_load($edit['uid'], TRUE);
 
-    $object = (object) $edit;
-    field_attach_insert('user', $object);
+      $object = (object) $edit;
+      field_attach_insert('user', $object);
 
-    user_module_invoke('insert', $edit, $user, $category);
+      user_module_invoke('insert', $edit, $user, $category);
 
-    // Note, we wait with saving the data column to prevent module-handled
-    // fields from being saved there.
-    $data = array();
-    foreach ($edit as $key => $value) {
-      // Form fields that don't pertain to the users, user_roles, or
-      // Field API are automatically serialized into the user.data
-      // column.
-      if ((!in_array($key, array('roles', 'is_new'))) && (empty($user_fields[$key]) && empty($field_form[$key])) && ($value !== NULL)) {
-        $data[$key] = $value;
+      // Note, we wait with saving the data column to prevent module-handled
+      // fields from being saved there.
+      $data = array();
+      foreach ($edit as $key => $value) {
+        // Form fields that don't pertain to the users, user_roles, or
+        // Field API are automatically serialized into the user.data
+        // column.
+        if ((!in_array($key, array('roles', 'is_new'))) && (empty($user_fields[$key]) && empty($field_form[$key])) && ($value !== NULL)) {
+          $data[$key] = $value;
+        }
+      }
+      if (!empty($data)) {
+        $data_array = array('uid' => $user->uid, 'data' => $data);
+        drupal_write_record('users', $data_array, 'uid');
       }
-    }
-    if (!empty($data)) {
-      $data_array = array('uid' => $user->uid, 'data' => $data);
-      drupal_write_record('users', $data_array, 'uid');
-    }
 
-    // Save user roles (delete just to be safe).
-    if (isset($edit['roles']) && is_array($edit['roles'])) {
-      db_delete('users_roles')
-        ->condition('uid', $edit['uid'])
-        ->execute();
-      $query = db_insert('users_roles')->fields(array('uid', 'rid'));
-      foreach (array_keys($edit['roles']) as $rid) {
-        if (!in_array($rid, array(DRUPAL_ANONYMOUS_RID, DRUPAL_AUTHENTICATED_RID))) {
-          $query->values(array(
-            'uid' => $edit['uid'],
-            'rid' => $rid,
-          ));
+      // Save user roles (delete just to be safe).
+      if (isset($edit['roles']) && is_array($edit['roles'])) {
+        db_delete('users_roles')
+          ->condition('uid', $edit['uid'])
+          ->execute();
+        $query = db_insert('users_roles')->fields(array('uid', 'rid'));
+        foreach (array_keys($edit['roles']) as $rid) {
+          if (!in_array($rid, array(DRUPAL_ANONYMOUS_RID, DRUPAL_AUTHENTICATED_RID))) {
+            $query->values(array(
+              'uid' => $edit['uid'],
+              'rid' => $rid,
+            ));
+          }
         }
+        $query->execute();
       }
-      $query->execute();
+
+      // Build the finished user object.
+      $user = user_load($edit['uid'], TRUE);
     }
 
-    // Build the finished user object.
-    $user = user_load($edit['uid'], TRUE);
+    return $user;
+  }
+  catch (Exception $e) {
+    $transaction->rollback('user', $e->getMessage(), array(), WATCHDOG_ERROR);
   }
-
-  return $user;
 }
 
 /**