Index: trunk/Nebulous-Server/Changes
===================================================================
--- trunk/Nebulous-Server/Changes	(revision 23455)
+++ trunk/Nebulous-Server/Changes	(revision 23458)
@@ -1,3 +1,6 @@
 Revision history for Nebulous
+
+0.17
+    - retry database transactions when a deadlock is detected
 
 0.16
Index: trunk/Nebulous-Server/lib/Nebulous/Server.pm
===================================================================
--- trunk/Nebulous-Server/lib/Nebulous/Server.pm	(revision 23455)
+++ trunk/Nebulous-Server/lib/Nebulous/Server.pm	(revision 23458)
@@ -163,73 +163,79 @@
 
     my $uri;
-    eval {
-        {
-            # create storage_object
-            my $query = $db->prepare_cached( $sql->new_object ); 
-            $query->execute('NULL', $key->path);
-        }
-
-        my $so_id;
-        {
-            # get object ID
-            my $query = $db->prepare_cached( $sql->last_insert_id );
-            $query->execute;
-            ($so_id) = $query->fetchrow_array;
-            # XXX finish seems to be required when using LAST_INSERT_ID() or we
-            # get a warning about the stmt handling still be active the next
-            # time LAST_INSERT_ID() is invoked
-            $query->finish;
-        }
-
-        {
-            # create storage_object_attr
-            my $query = $db->prepare_cached( $sql->new_object_attr ); 
-            $query->execute($so_id);
-        }
-
-        {
-            
-            # create instance with no URI
+TRANS: do {
+        eval {
+            {
+                # create storage_object
+                my $query = $db->prepare_cached( $sql->new_object ); 
+                $query->execute('NULL', $key->path);
+            }
+
+            my $so_id;
+            {
+                # get object ID
+                my $query = $db->prepare_cached( $sql->last_insert_id );
+                $query->execute;
+                ($so_id) = $query->fetchrow_array;
+                # XXX finish seems to be required when using LAST_INSERT_ID() or we
+                # get a warning about the stmt handling still be active the next
+                # time LAST_INSERT_ID() is invoked
+                $query->finish;
+            }
+
+            {
+                # create storage_object_attr
+                my $query = $db->prepare_cached( $sql->new_object_attr ); 
+                $query->execute($so_id);
+            }
+
+            {
+                
+                # create instance with no URI
 #            my $query = $db->prepare_cached( $sql->new_instance );
-            my $query = $db->prepare_cached( $sql->new_object_instance );
-            $query->execute($vol_id);
-        }
-
-        my $ins_id;
-        {
-            # get instance ID
-            my $query = $db->prepare_cached( $sql->last_insert_id );
-            $query->execute;
-            ($ins_id) = $query->fetchrow_array;
-            # XXX finish seems to be required when using LAST_INSERT_ID() or we
-            # get a warning about the stmt handling still be active the next
-            # time LAST_INSERT_ID() is invoked
-            $query->finish;
-        }
-
-        # Unfortunately, since we want to use the instance row's ID as part of the
-        # actual on disk file name we can't try to create the file until after
-        # we've create both a new storage_storage object and instance.
-
-        # TODO add some stuff here to retry if unsucessful
-        $uri = $self->_create_empty_instance_file($key->path, $so_id, $ins_id, $vol_path, $vol_xattr);
-        $log->debug("created $uri on volume ID: $vol_id");
-
-        {
-            # update the instance with URI & vol_id that the file is on
-            my $query = $db->prepare_cached( $sql->update_instance_uri );
-            # vol_id, uri, ins_id
-            $query->execute($vol_id, "$uri", $ins_id);
-        }
-
-        $db->commit;
-        $log->debug("commit");
-    };
-    if ($@) {
+                my $query = $db->prepare_cached( $sql->new_object_instance );
+                $query->execute($vol_id);
+            }
+
+            my $ins_id;
+            {
+                # get instance ID
+                my $query = $db->prepare_cached( $sql->last_insert_id );
+                $query->execute;
+                ($ins_id) = $query->fetchrow_array;
+                # XXX finish seems to be required when using LAST_INSERT_ID() or we
+                # get a warning about the stmt handling still be active the next
+                # time LAST_INSERT_ID() is invoked
+                $query->finish;
+            }
+
+            # Unfortunately, since we want to use the instance row's ID as part of the
+            # actual on disk file name we can't try to create the file until after
+            # we've create both a new storage_storage object and instance.
+
+            # TODO add some stuff here to retry if unsucessful
+            $uri = $self->_create_empty_instance_file($key->path, $so_id, $ins_id, $vol_path, $vol_xattr);
+            $log->debug("created $uri on volume ID: $vol_id");
+
+            {
+                # update the instance with URI & vol_id that the file is on
+                my $query = $db->prepare_cached( $sql->update_instance_uri );
+                # vol_id, uri, ins_id
+                $query->execute($vol_id, "$uri", $ins_id);
+            }
+
+            $db->commit;
+            $log->debug("commit");
+        };
+        if ($@) {
 #        and not $key->soft_volume
-        $db->rollback;
-        $log->debug("rollback");
-        $log->logdie("error: $@");
-    }
+            $db->rollback;
+            $log->debug("rollback");
+            if ($@ =~ /Deadlock found/) {
+                $log->warn("database deadlock retrying transaction: $@");
+                redo TRANS;
+            }
+            $log->logdie("error: $@");
+        }
+    } while ($@);
 
     $log->debug("leaving");
@@ -269,5 +275,5 @@
     $newkey = parse_neb_key($newkey);
 
-    TRANS: do {
+TRANS: do {
         eval {
             # rename storage_object
@@ -290,5 +296,5 @@
             if ($@ =~ /Deadlock found/) {
                 $log->warn("database deadlock retrying transaction: $@");
-                redo;
+                redo TRANS;
             }
             $log->logdie("database error: $@");
@@ -336,52 +342,58 @@
     # key1.swap -> key2
 
-    eval {
-        {
-            # key1 -> key1.swap
-            my $query = $db->prepare_cached($sql->rename_object); 
-            # this SQL statment takes the new key name as the first param
-            my $rows = $query->execute($key1->path . ".swap", $key1->path);
-
-            # if we affected more then one row something very bad has happened.
-            unless ($rows == 1) {
-                $query->finish;
-                $log->logdie("affected row count is $rows instead of 1");
-            }
-        }
-
-        {
-            # key2 -> key1
-            my $query = $db->prepare_cached($sql->rename_object); 
-            # this SQL statment takes the new key name as the first param
-            my $rows = $query->execute($key1->path, $key2->path);
-
-            # if we affected more then one row something very bad has happened.
-            unless ($rows == 1) {
-                $query->finish;
-                $log->logdie("affected row count is $rows instead of 1");
-            }
-        }
-
-        {
-            # key1.swap -> key2
-            my $query = $db->prepare_cached($sql->rename_object); 
-            # this SQL statment takes the new key name as the first param
-            my $rows = $query->execute($key2->path, $key1->path . ".swap");
-
-            # if we affected more then one row something very bad has happened.
-            unless ($rows == 1) {
-                $query->finish;
-                $log->logdie("affected row count is $rows instead of 1");
-            }
-        }
-
-        $db->commit;
-        $log->debug("commit");
-    };
-    if ($@) {
-        $db->rollback;
-        $log->debug("rollback");
-        $log->logdie("database error: $@");
-    }
+TRANS: do {
+        eval {
+            {
+                # key1 -> key1.swap
+                my $query = $db->prepare_cached($sql->rename_object); 
+                # this SQL statment takes the new key name as the first param
+                my $rows = $query->execute($key1->path . ".swap", $key1->path);
+
+                # if we affected more then one row something very bad has happened.
+                unless ($rows == 1) {
+                    $query->finish;
+                    $log->logdie("affected row count is $rows instead of 1");
+                }
+            }
+
+            {
+                # key2 -> key1
+                my $query = $db->prepare_cached($sql->rename_object); 
+                # this SQL statment takes the new key name as the first param
+                my $rows = $query->execute($key1->path, $key2->path);
+
+                # if we affected more then one row something very bad has happened.
+                unless ($rows == 1) {
+                    $query->finish;
+                    $log->logdie("affected row count is $rows instead of 1");
+                }
+            }
+
+            {
+                # key1.swap -> key2
+                my $query = $db->prepare_cached($sql->rename_object); 
+                # this SQL statment takes the new key name as the first param
+                my $rows = $query->execute($key2->path, $key1->path . ".swap");
+
+                # if we affected more then one row something very bad has happened.
+                unless ($rows == 1) {
+                    $query->finish;
+                    $log->logdie("affected row count is $rows instead of 1");
+                }
+            }
+
+            $db->commit;
+            $log->debug("commit");
+        };
+        if ($@) {
+            $db->rollback;
+            $log->debug("rollback");
+            if ($@ =~ /Deadlock found/) {
+                $log->warn("database deadlock retrying transaction: $@");
+                redo TRANS;
+            }
+            $log->logdie("database error: $@");
+        }
+    } while ($@);
 
     $log->debug("leaving");
@@ -454,62 +466,64 @@
 
     my $uri;
-    eval {
-        my $so_id;
-        {
-            # verify that at least one instance is currently available
-            my $query = $db->prepare_cached( $sql->get_object_instances );
-            my $rows = $query->execute($key->path, 1);
-
-            unless ( $rows > 0 ) {
+TRANS: do {
+        eval {
+            my $so_id;
+            {
+                # verify that at least one instance is currently available
+                my $query = $db->prepare_cached( $sql->get_object_instances );
+                my $rows = $query->execute($key->path, 1);
+
+                unless ( $rows > 0 ) {
+                    $query->finish;
+                    $log->logdie( "storage object does not exist" );
+                }
+
+                $so_id = $query->fetchrow_hashref->{ 'so_id' };
                 $query->finish;
-                $log->logdie( "storage object does not exist" );
-            }
-
-            $so_id = $query->fetchrow_hashref->{ 'so_id' };
-            $query->finish;
-        }
-
-        {
-            my $query = $db->prepare_cached( $sql->new_instance );
-            $query->execute($so_id, $vol_id);
-        }
-
-        my $ins_id;
-        {
-            my $query = $db->prepare_cached( $sql->last_insert_id );
-            $query->execute();
-            ($ins_id) = $query->fetchrow_array;
-            # XXX finish seems to be required when using LAST_INSERT_ID() or we
-            # get a warning about the stmt handling still being active the next
-            # time LAST_INSERT_ID() is invoked
-            $query->finish;
-        }
-
-        # Unfortunately, since we want to use the instance row's ID as part of
-        # the actual on disk file name we can't try to create the file until
-        # after we've create both a new storage_storage object and instance.
-
-        # TODO add some stuff here to retry if unsucessful
-        $uri = $self->_create_empty_instance_file($key->path, $so_id, $ins_id, $vol_path, $vol_xattr);
-
-        {
-            my $query = $db->prepare_cached( $sql->update_instance_uri );
-            # vol_id, uri, ins_id
-            $query->execute($vol_id, $uri, $ins_id);
-        }
-
-        $db->commit;
-        $log->debug("commit");
-    };
-    if ($@) {
-        $db->rollback;
-        # handle soft volumes
-        if (defined $vol_name and defined $key->soft_volume) {
-            $log->debug("retrying with 'any' volume");
-            return $self->replicate_object($key->path, 'any');
-        }
-        $log->debug("rollback");
-        $log->logdie("error: $@");
-    }
+            }
+
+            {
+                my $query = $db->prepare_cached( $sql->new_instance );
+                $query->execute($so_id, $vol_id);
+            }
+
+            my $ins_id;
+            {
+                my $query = $db->prepare_cached( $sql->last_insert_id );
+                $query->execute();
+                ($ins_id) = $query->fetchrow_array;
+                # XXX finish seems to be required when using LAST_INSERT_ID() or we
+                # get a warning about the stmt handling still being active the next
+                # time LAST_INSERT_ID() is invoked
+                $query->finish;
+            }
+
+            # Unfortunately, since we want to use the instance row's ID as part of
+            # the actual on disk file name we can't try to create the file until
+            # after we've create both a new storage_storage object and instance.
+
+            # TODO add some stuff here to retry if unsucessful
+            $uri = $self->_create_empty_instance_file($key->path, $so_id, $ins_id, $vol_path, $vol_xattr);
+
+            {
+                my $query = $db->prepare_cached( $sql->update_instance_uri );
+                # vol_id, uri, ins_id
+                $query->execute($vol_id, $uri, $ins_id);
+            }
+
+            $db->commit;
+            $log->debug("commit");
+        };
+        if ($@) {
+            $db->rollback;
+            # handle soft volumes
+            if (defined $vol_name and defined $key->soft_volume) {
+                $log->debug("retrying with 'any' volume");
+                return $self->replicate_object($key->path, 'any');
+            }
+            $log->debug("rollback");
+            $log->logdie("error: $@");
+        }
+    } while ($@);
 
     $log->debug("leaving");
@@ -551,68 +565,74 @@
     my $write_lock;
 
-    eval {
-        {
-            # this will set update locks
-            my $query = $db->prepare_cached( $sql->get_object_locks );
-            my $rows = $query->execute( $key->path );
-            unless ( $rows == 1 ) {
+TRANS: do {
+        eval {
+            {
+                # this will set update locks
+                my $query = $db->prepare_cached( $sql->get_object_locks );
+                my $rows = $query->execute( $key->path );
+                unless ( $rows == 1 ) {
+                    $query->finish;
+                    $log->logdie( "storage object does not exist" );
+                }
+
+                my $row = $query->fetchrow_hashref;
                 $query->finish;
-                $log->logdie( "storage object does not exist" );
-            }
-
-            my $row = $query->fetchrow_hashref;
-            $query->finish;
-
-            $so_id      = $row->{ 'so_id' };
-            $read_lock  = $row->{ 'read_lock' };
-            $write_lock = $row->{ 'write_lock' };
-        }
-
-        if ($type eq 'write') {
-            # can't set a write lock twice and
-            # can't set a write lock if there are read locks
-            if ($write_lock) {
-                $log->logdie("can not write lock twice -- retry");
-            }
-            
-            if ($read_lock > 0) {
-                $log->logdie("can not write lock after read lock -- retry");
-            }
-
-            {
-                my $query = $db->prepare_cached( $sql->set_write_lock );
-                my $rows = $query->execute($key->path);
-            
-                # if we affected more then one row something very bad has happened.
-                unless ($rows == 1) {
-                    $log->logdie("affected row count is $rows instead of 1");
-                }
-
-            }
-        } elsif ($type eq 'read') {
-            # can't set a read lock if there's a write lock
-            if ($write_lock) {
-                $log->logdie("can not read lock after write lock -- retry");
-            }
-
-            {
-                my $query = $db->prepare_cached( $sql->increment_read_lock );
-                my $rows = $query->execute($key->path);
-            
-                # if we affected more then one row something very bad has happened.
-                unless ($rows == 1) {
-                    $log->logdie("affected row count is $rows instead of 1");
-                }
-            }
-        }
-
-        $db->commit;
-        $log->debug("commit");
-    };
-    if ($@) {
-        $db->rollback;
-        $log->debug("rollback");
-        $log->logdie("error: $@");
-    }
+
+                $so_id      = $row->{ 'so_id' };
+                $read_lock  = $row->{ 'read_lock' };
+                $write_lock = $row->{ 'write_lock' };
+            }
+
+            if ($type eq 'write') {
+                # can't set a write lock twice and
+                # can't set a write lock if there are read locks
+                if ($write_lock) {
+                    $log->logdie("can not write lock twice -- retry");
+                }
+                
+                if ($read_lock > 0) {
+                    $log->logdie("can not write lock after read lock -- retry");
+                }
+
+                {
+                    my $query = $db->prepare_cached( $sql->set_write_lock );
+                    my $rows = $query->execute($key->path);
+                
+                    # if we affected more then one row something very bad has happened.
+                    unless ($rows == 1) {
+                        $log->logdie("affected row count is $rows instead of 1");
+                    }
+
+                }
+            } elsif ($type eq 'read') {
+                # can't set a read lock if there's a write lock
+                if ($write_lock) {
+                    $log->logdie("can not read lock after write lock -- retry");
+                }
+
+                {
+                    my $query = $db->prepare_cached( $sql->increment_read_lock );
+                    my $rows = $query->execute($key->path);
+                
+                    # if we affected more then one row something very bad has happened.
+                    unless ($rows == 1) {
+                        $log->logdie("affected row count is $rows instead of 1");
+                    }
+                }
+            }
+
+            $db->commit;
+            $log->debug("commit");
+        };
+        if ($@) {
+            $db->rollback;
+            $log->debug("rollback");
+            if ($@ =~ /Deadlock found/) {
+                $log->warn("database deadlock retrying transaction: $@");
+                redo TRANS;
+            }
+            $log->logdie("error: $@");
+        }
+    } while ($@);
 
     $log->debug("leaving");
@@ -654,71 +674,77 @@
     my $write_lock;
 
-    eval {
-        {
-            # this will set update locks
-            my $query = $db->prepare_cached( $sql->get_object_locks );
-            my $rows = $query->execute($key->path);
-            unless ($rows == 1) {
+TRANS: do {
+        eval {
+            {
+                # this will set update locks
+                my $query = $db->prepare_cached( $sql->get_object_locks );
+                my $rows = $query->execute($key->path);
+                unless ($rows == 1) {
+                    $query->finish;
+                    $log->logdie("storage object does not exist");
+                }
+
+                my $row = $query->fetchrow_hashref;
                 $query->finish;
-                $log->logdie("storage object does not exist");
-            }
-
-            my $row = $query->fetchrow_hashref;
-            $query->finish;
-
-            $so_id      = $row->{ 'so_id' };
-            $read_lock  = $row->{ 'read_lock' };
-            $write_lock = $row->{ 'write_lock' };
-        }
-
-        if ($type eq 'write') {
-            # can't remove a write lock if it doesn't exist
-            if ($read_lock) {
-                $log->logdie("can not have a write lock under a read lock");
-            }
-
-            unless ($write_lock) {
-                $log->logdie("can not remove non-existant write lock");
-            }
-
-            {
-                my $query = $db->prepare_cached( $sql->delete_write_lock );
-                my $rows = $query->execute($key->path);
-            
-                # if we affected more then one row something very bad has happened.
-                unless ($rows == 1) {
-                    $log->logdie("affected row count is $rows instead of 1");
-                }
-            }
-        } elsif ($type eq 'read') {
-            # can't remove a read lock if there's a write lock and
-            # can't remove a read lock if there aren't any
-            if ($write_lock) {
-                $log->logdie("can not have a read lock under a write lock");
-            }
-               
-            if ($read_lock == 0) {
-                $log->logdie("can not remove non-existant read lock");
-            }
-
-            {
-                my $query = $db->prepare_cached( $sql->decrement_read_lock );
-                my $rows = $query->execute($key->path);
-            
-                # if we affected more then one row something very bad has happened.
-                unless ($rows == 1) {
-                    $log->logdie("affected row count is $rows instead of 1");
-                }
-
-            }
-        }
-        $db->commit;
-        $log->debug("commit");
-    };
-    if ($@) {
-        $db->rollback;
-        $log->debug("rollback");
-        $log->logdie("error: $@");
-    }
+
+                $so_id      = $row->{ 'so_id' };
+                $read_lock  = $row->{ 'read_lock' };
+                $write_lock = $row->{ 'write_lock' };
+            }
+
+            if ($type eq 'write') {
+                # can't remove a write lock if it doesn't exist
+                if ($read_lock) {
+                    $log->logdie("can not have a write lock under a read lock");
+                }
+
+                unless ($write_lock) {
+                    $log->logdie("can not remove non-existant write lock");
+                }
+
+                {
+                    my $query = $db->prepare_cached( $sql->delete_write_lock );
+                    my $rows = $query->execute($key->path);
+                
+                    # if we affected more then one row something very bad has happened.
+                    unless ($rows == 1) {
+                        $log->logdie("affected row count is $rows instead of 1");
+                    }
+                }
+            } elsif ($type eq 'read') {
+                # can't remove a read lock if there's a write lock and
+                # can't remove a read lock if there aren't any
+                if ($write_lock) {
+                    $log->logdie("can not have a read lock under a write lock");
+                }
+                   
+                if ($read_lock == 0) {
+                    $log->logdie("can not remove non-existant read lock");
+                }
+
+                {
+                    my $query = $db->prepare_cached( $sql->decrement_read_lock );
+                    my $rows = $query->execute($key->path);
+                
+                    # if we affected more then one row something very bad has happened.
+                    unless ($rows == 1) {
+                        $log->logdie("affected row count is $rows instead of 1");
+                    }
+
+                }
+            }
+            $db->commit;
+            $log->debug("commit");
+        };
+        if ($@) {
+            $db->rollback;
+            $log->debug("rollback");
+            if ($@ =~ /Deadlock found/) {
+                $log->warn("database deadlock retrying transaction: $@");
+                redo TRANS;
+            }
+            $log->logdie("error: $@");
+        }
+    } while ($@);
 
     $log->debug( "leaving" );
@@ -762,39 +788,45 @@
     $key = parse_neb_key($key);
 
-    eval {
-        my $query;
-
-        if ($flags eq 'create') {
-            $query = $db->prepare_cached( $sql->new_object_xattr );
-        } else {
-            # replace
-            $query = $db->prepare_cached( $sql->replace_object_xattr );
-        }
-
-        # name, value, ext_id
-        my $rows = $query->execute($name, $value, $key->path);
-        $query->finish;
-
-        # if we affected more then one row something very bad has happened.
-        if ($flags eq 'create') {
-            unless ($rows == 1) {
-                $log->logdie( "affected row count is $rows instead of 1" );
-            }
-        } else {
-            # replace_object_xattr can effect either 1 or 2 rows.  2 rows in
-            # the case of a replace and 1 if the xattr didn't already exist.
-            unless ($rows == 1 or $rows == 2) {
-                $log->logdie( "affected row count is $rows instead of 2" );
-            }
-        }
-
-        $db->commit;
-        $log->debug("commit");
-    };
-    if ($@) {
-        $db->rollback;
-        $log->debug("rollback");
-        $log->logdie("database error: $@");
-    }
+TRANS: do {
+        eval {
+            my $query;
+
+            if ($flags eq 'create') {
+                $query = $db->prepare_cached( $sql->new_object_xattr );
+            } else {
+                # replace
+                $query = $db->prepare_cached( $sql->replace_object_xattr );
+            }
+
+            # name, value, ext_id
+            my $rows = $query->execute($name, $value, $key->path);
+            $query->finish;
+
+            # if we affected more then one row something very bad has happened.
+            if ($flags eq 'create') {
+                unless ($rows == 1) {
+                    $log->logdie( "affected row count is $rows instead of 1" );
+                }
+            } else {
+                # replace_object_xattr can effect either 1 or 2 rows.  2 rows in
+                # the case of a replace and 1 if the xattr didn't already exist.
+                unless ($rows == 1 or $rows == 2) {
+                    $log->logdie( "affected row count is $rows instead of 2" );
+                }
+            }
+
+            $db->commit;
+            $log->debug("commit");
+        };
+        if ($@) {
+            $db->rollback;
+            $log->debug("rollback");
+            if ($@ =~ /Deadlock found/) {
+                $log->warn("database deadlock retrying transaction: $@");
+                redo TRANS;
+            }
+            $log->logdie("database error: $@");
+        }
+    } while ($@);
 
     $log->debug("leaving");
@@ -925,23 +957,29 @@
     $key = parse_neb_key($key);
 
-    eval {
-        my $query = $db->prepare_cached( $sql->remove_object_xattr );
-        # ext_id, name
-        my $rows = $query->execute($key->path, $name);
-        $query->finish;
-
-        # if we affected more then one row something very bad has happened.
-        unless ($rows == 1) {
-            $log->logdie( "affected row count is $rows instead of 1" );
-        }
-
-        $db->commit;
-        $log->debug("commit");
-    };
-    if ($@) {
-        $db->rollback;
-        $log->debug("rollback");
-        $log->logdie("database error: $@");
-    }
+TRANS: do {
+        eval {
+            my $query = $db->prepare_cached( $sql->remove_object_xattr );
+            # ext_id, name
+            my $rows = $query->execute($key->path, $name);
+            $query->finish;
+
+            # if we affected more then one row something very bad has happened.
+            unless ($rows == 1) {
+                $log->logdie( "affected row count is $rows instead of 1" );
+            }
+
+            $db->commit;
+            $log->debug("commit");
+        };
+        if ($@) {
+            $db->rollback;
+            $log->debug("rollback");
+            if ($@ =~ /Deadlock found/) {
+                $log->warn("database deadlock retrying transaction: $@");
+                redo TRANS;
+            }
+            $log->logdie("database error: $@");
+        }
+    } while ($@);
 
     $log->debug("leaving");
@@ -1105,65 +1143,71 @@
     $log->debug( "entered - @_" );
 
-    eval {
-        my $so_id;
-        my $instances;
-        # get so_id
-        {
-            my $query = $db->prepare_cached( $sql->get_object_from_uri );
-            my $rows = $query->execute( $uri );
-
-            unless ( $rows > 0 ) {
+TRANS: do {
+        eval {
+            my $so_id;
+            my $instances;
+            # get so_id
+            {
+                my $query = $db->prepare_cached( $sql->get_object_from_uri );
+                my $rows = $query->execute( $uri );
+
+                unless ( $rows > 0 ) {
+                    $query->finish;
+                    $log->logdie( "no instance is associated with uri" );
+                }
+
+                $so_id = $query->fetchrow_hashref->{ 'so_id' };
                 $query->finish;
-                $log->logdie( "no instance is associated with uri" );
-            }
-
-            $so_id = $query->fetchrow_hashref->{ 'so_id' };
-            $query->finish;
-
-        }
-
-        {
-            my $query = $db->prepare_cached( $sql->get_instance_count );
-            $query->execute( $so_id );
-
-            $instances = $query->fetchrow_hashref->{ 'count(ins_id)' };
-            $query->finish;
-        }
-
-        # remove instance
-        {
-            my $query = $db->prepare_cached( $sql->delete_instance );
-            my $rows = $query->execute( $uri );
-            $query->finish;
-            
-            # if we affected something other then two rows something very bad
-            # has happened
-            unless ( $rows == 1 ) {
-                $log->logdie( "affected row count is $rows instead of 1" );
-            }
-        }
-
-        # if we just deleted the last instance associated with a storage object
-        # remove it too
-        if ( $instances == 1 ) {
-            # we just removed the last instance
-            my $query = $db->prepare_cached( $sql->delete_object );
-            my $rows = $query->execute( $so_id );
-            $query->finish;
-
-            # TODO: this will have to be changed in order to support hardlinks
-            unless ( $rows == 1 ) {
-                $log->logdie( "affected row count is $rows instead of 2" );
-            }
-        }
-
-        $db->commit;
-        $log->debug("commit");
-    };
-    if ( $@ ) {
-        $db->rollback;
-        $log->debug("rollback");
-        $log->logdie( "database error: $@" );
-    }
+
+            }
+
+            {
+                my $query = $db->prepare_cached( $sql->get_instance_count );
+                $query->execute( $so_id );
+
+                $instances = $query->fetchrow_hashref->{ 'count(ins_id)' };
+                $query->finish;
+            }
+
+            # remove instance
+            {
+                my $query = $db->prepare_cached( $sql->delete_instance );
+                my $rows = $query->execute( $uri );
+                $query->finish;
+                
+                # if we affected something other then two rows something very bad
+                # has happened
+                unless ( $rows == 1 ) {
+                    $log->logdie( "affected row count is $rows instead of 1" );
+                }
+            }
+
+            # if we just deleted the last instance associated with a storage object
+            # remove it too
+            if ( $instances == 1 ) {
+                # we just removed the last instance
+                my $query = $db->prepare_cached( $sql->delete_object );
+                my $rows = $query->execute( $so_id );
+                $query->finish;
+
+                # TODO: this will have to be changed in order to support hardlinks
+                unless ( $rows == 1 ) {
+                    $log->logdie( "affected row count is $rows instead of 2" );
+                }
+            }
+
+            $db->commit;
+            $log->debug("commit");
+        };
+        if ( $@ ) {
+            $db->rollback;
+            $log->debug("rollback");
+            if ($@ =~ /Deadlock found/) {
+                $log->warn("database deadlock retrying transaction: $@");
+                redo TRANS;
+            }
+            $log->logdie( "database error: $@" );
+        }
+    } while ($@);
 
     $log->debug( "leaving" );