Changeset 24557 for branches/eam_branches/20090522/Nebulous-Server/lib/Nebulous/Server.pm

Timestamp:

Jun 25, 2009, 2:00:56 PM (17 years ago)

Author:

eugene

Message:

merging changes from head

Location:

branches/eam_branches/20090522

Files:

• . (modified) (1 prop)
• Nebulous-Server (modified) (1 prop)
• Nebulous-Server/lib/Nebulous/Server.pm (modified) (109 diffs)

branches/eam_branches/20090522/Nebulous-Server/lib/Nebulous/Server.pm

@@ -13 +13 @@
 use base qw( Class::Accessor::Fast );
 
+use Cache::Memcached;
 use DBI;
 use Digest::SHA1 qw( sha1_hex );
+use Fcntl ':mode';
 use File::Basename qw( basename dirname fileparse );
 use File::ExtAttr qw( setfattr );
@@ -27 +29 @@
 use URI::file;
 
-__PACKAGE__->mk_accessors(qw( log sql config ));
-
-use constant SUBPATH_DEPTH  => 2;
-use constant NFS_RETRIES    => 100;
+__PACKAGE__->mk_accessors(qw( log sql config cache ));
+
+use constant SUBPATH_DEPTH      => 2;
+use constant NFS_RETRIES        => 100;
+use constant NFS_RETRY_WAIT     => 1;
+use constant TRANS_RETRY_WAIT   => 1;
+
+# transaction restart/retry regex
+my $trans_regex = qr/Deadlock Found|Lock wait timeout exceeded|try restarting transaction|Can't connect to MySQL server/i;
 
 sub new
@@ -45 +52 @@
 sub new_from_config
 {
-    my ($class, $config) = @_;
+    my $class = shift;
+
+    my ($config) = @_;
 
     # log4perl is not avaliable until we call init()
@@ -51 +60 @@
     my $log = Log::Log4perl::get_logger( "Nebulous::Server" );
     $log->level($config->trace);
+    $log->debug( "entered - @_" );
 
     my $sql = Nebulous::Server::SQL->new;
-
-    $log->debug( "entered - @_" );
 
     my $self = bless {}, ref $class || $class;
@@ -60 +68 @@
     $self->sql($sql);
     $self->config($config);
+    $self->cache(
+        Cache::Memcached->new({
+            servers => $config->memcached_servers,
+       })
+    );
+#    $self->cache->set("foo", "bar") or die "set failed";
+#    $self->cache->get("foo") or die "get failed";
+    $log->logdie("at least one database must be defined") unless $config->n_db;
+
+    # cause a db session to be started
+    $self->_db_for_index(0);
 
     $log->debug( "leaving" );
@@ -70 +89 @@
 sub _db_index_for_key
 {
-    my ($self, $key) = @_;
+    my $self = shift;
+
+    my $log     = $self->log;
+    $log->debug( "entered - @_" );
+
+    my ($key) = @_;
 
     my $config  = $self->config;
 
     my $db_index = 0;
-    die "key not defined" unless defined $key;
+    $log->logdie("key not defined") unless defined $key;
 
     # hash the key to select the correct database instance
@@ -85 +109 @@
     $db_index = unpack("h8", sha1_hex($path)) % $config->n_db;
 
+    $log->debug("index is $db_index");
+    $log->debug("leaving");
+
     return $db_index;
 }
@@ -90 +117 @@
 sub _db_for_index
 {
-    my ($self, $db_index) = @_;
+    my $self = shift;
 
     my $log     = $self->log;
+    $log->debug( "entered - @_" );
+
+    my ($db_index) = @_;
+
     my $sql     = $self->sql;
     my $config  = $self->config;
@@ -108 +139 @@
     # lookup database info
     my $db_config = $config->db($db_index);
-    die "can't find database configuration info for database # $db_index"
+    $log->logdie("can't find database configuration info for db # $db_index")
         unless $db_config;
 
@@ -116 +147 @@
     # processes and the database might have gone away on us.  Apache::DBI will
     # take care of getting a valid dbh back.
-    eval {
-        $dbh = DBI->connect_cached(
-            $db_config->dsn,
-            $db_config->dbuser,
-            $db_config->dbpasswd,
-            {
-                RaiseError => 1,
-                PrintError => 0,
-                AutoCommit => 0,
-            },
-        );
-
-        $dbh->do( $sql->set_transaction_model );
-        $log->debug( "connected to database: ", sub { $dbh->data_sources; } );
-        $dbh->commit;
-        $log->debug("commit");
-    };
-    if ( $@ ) {
-        $dbh->rollback if $dbh;
-        $log->debug("rollback");
-        $log->logdie( "database error: $@" );
+    TRANS: while (1) {
+        eval {
+            $dbh = DBI->connect_cached(
+                $db_config->dsn,
+                $db_config->dbuser,
+                $db_config->dbpasswd,
+                {
+                    RaiseError => 1,
+                    PrintError => 0,
+                    AutoCommit => 0,
+                },
+            );
+
+            $dbh->do( $sql->set_transaction_model );
+            $log->debug( "connected to database: ", sub { $dbh->data_sources; } );
+            $dbh->commit;
+            $log->debug("commit");
+        };
+        if ($@) {
+            $dbh->rollback if $dbh;
+            $log->debug("rollback") if $dbh;
+            if ($@ =~ qr/Can't connect to MySQL server/) {
+                $log->warn("database error, retrying transaction: $@");
+                sleep TRANS_RETRY_WAIT;
+                redo TRANS;
+            }
+            $log->logdie( "database error: $@" );
+        }
+        last;
     }
 
     $self->{dbs}[$db_index] = $dbh;
 
+    $log->debug("leaving");
+
     return $dbh;
 }
@@ -148 +189 @@
     my $self = shift;
 
+    my $log     = $self->log;
+    $log->debug( "entered - @_" );
+
     my ($key) = validate_pos(@_,
         {
@@ -154 +198 @@
     );
 
-    my $log     = $self->log;
     my $sql     = $self->sql;
     my $config  = $self->config;
 
-    die "key not defined" unless defined $key;
+    $log->logdie("key not defined") unless defined $key;
     my $db_index = $self->_db_index_for_key($key);
 
     my $dbh = $self->_db_for_index($db_index);
+
     return $dbh;
 }
@@ -168 +212 @@
 {
     my $self = shift;
+
+    my $log = $self->log;
+    $log->debug( "entered - @_" );
 
     my ($key, $vol_name) = validate_pos(@_,
@@ -186 +233 @@
     );
 
+    my $sql = $self->sql;
+
     # vol_name overrides the key implied volume
-    $key = parse_neb_key($key, $vol_name);
+    eval {
+        $key = parse_neb_key($key, $vol_name);
+    };
+    $log->logdie("$@") if $@;
     $vol_name = $key->volume;
 
-    my $log = $self->log;
-    my $sql = $self->sql;
     my $db  = $self->db($key);
-
-    $log->debug( "entered - @_" );
 
     # the key's volume can't be validiated on input for this method so we have
@@ -200 +248 @@
     if (defined $vol_name
         and not $self->_is_valid_volume_name($key, $key->volume)) {
-        if ($key->soft_volume) {
+        unless ($key->hard_volume) {
             $log->warn( "$vol_name is not a known volume name" );
             $vol_name = undef;
         } else {
-            die "$vol_name is not a valid volume name"
+           $log->logdie("$vol_name is not a valid volume name");
         }
     }
         
     my ($vol_id, $vol_host, $vol_path, $vol_xattr)
-        = $self->_get_storage_volume($key, $vol_name, $key->soft_volume);
+        = $self->_get_storage_volume($key, $vol_name, $key->hard_volume);
 
     my $parent_id = $self->_resolve_dir_parent_id(key => $key, create => 1);
@@ -282 +330 @@
             $db->rollback;
             $log->debug("rollback");
-            if ($@ =~ /Deadlock found/) {
-                $log->warn("database deadlock retrying transaction: $@");
+            if ($@ =~ $trans_regex) {
+                $log->warn("database error, retrying transaction: $@");
+                sleep TRANS_RETRY_WAIT;
                 redo TRANS;
             }
@@ -291 +340 @@
     }
 
+    # add new key to the cache
+    $self->cache->set($key->path, 1) if defined $self->cache;
+    $log->debug( "key added to cache" );
+
     $log->debug("leaving");
 
@@ -300 +353 @@
 {
     my $self = shift;
+
+    my $log = $self->log;
+    $log->debug( "entered - @_" );
 
     my %p = validate(@_,
@@ -322 +378 @@
     my $key = $p{key};
 
-    my $log = $self->log;
     my $sql = $self->sql;
     my $db  = $self->db($key);
-
-    $log->debug( "entered - @_" );
 
     # no path means '/', which has a dir_id & parent_id of 1
@@ -394 +447 @@
                     $query->finish;
                 }
-                $log->logdie("failed to get LAST_INSERT_ID()")
+                die("failed to get LAST_INSERT_ID()")
                     unless $parent_id;
 
-                $db->commit;
+#                $db->commit;
             }
         };
@@ -403 +456 @@
             $db->rollback;
             $log->debug("rollback");
-            if ($@ =~ /Deadlock found/) {
-                $log->warn("database deadlock retrying transaction: $@");
+            if ($@ =~ $trans_regex) {
+                $log->warn("database error, retrying transaction: $@");
+                $parent_id = 1;
+                sleep TRANS_RETRY_WAIT;
                 redo TRANS;
             }
+            if ($@ =~ qr/Duplicate entry/) {
+                $log->warn("Duplicate database entry, retrying transaction: $@");
+                $parent_id = 1;
+                sleep TRANS_RETRY_WAIT;
+                redo TRANS;
+            }
             $log->logdie("error: $@");
         }
@@ -421 +482 @@
 {
     my $self = shift;
+
+    my $log = $self->log;
+    $log->debug("entered - @_");
 
     my ($key, $newkey) = validate_pos(@_,
@@ -438 +502 @@
     );
 
+    my $sql = $self->sql;
+
     # ignore volumes
-    $key    = parse_neb_key($key);
-    $newkey = parse_neb_key($newkey);
-
-    my $log = $self->log;
-    my $sql = $self->sql;
+    eval {
+        $key = parse_neb_key($key);
+    };
+    $log->logdie("$@") if $@;
+    eval {
+        $newkey = parse_neb_key($newkey);
+    };
+    $log->logdie("$@") if $@;
+
     my $db  = $self->db($key);
-
-    $log->debug("entered - @_");
 
     # XXX this may require database migration in the future
@@ -464 +532 @@
             unless ($rows == 1) {
                 $query->finish;
-                $log->logdie("affected row count is $rows instead of 1");
-            }
+                die("affected row count is $rows instead of 1");
+            }
+
+            $self->cache->delete($key->path) if defined $self->cache;
+            $self->cache->set($newkey->path, 1) if defined $self->cache;
 
             $db->commit;
@@ -473 +544 @@
             $db->rollback;
             $log->debug("rollback");
-            if ($@ =~ /Deadlock found/) {
-                $log->warn("database deadlock retrying transaction: $@");
+            if ($@ =~ $trans_regex) {
+                $log->warn("database error, retrying transaction: $@");
+                sleep TRANS_RETRY_WAIT;
                 redo TRANS;
             }
@@ -491 +563 @@
     my $self = shift;
 
+    my $log = $self->log;
+    $log->debug("entered - @_");
+
     my ($key1, $key2) = validate_pos(@_,
         {
@@ -506 +581 @@
     );
 
+    my $sql = $self->sql;
+
     # ignore volumes
-    $key1 = parse_neb_key($key1);
-    $key2 = parse_neb_key($key2);
-
-    my $log = $self->log;
-    my $sql = $self->sql;
+    eval {
+        $key1 = parse_neb_key($key1);
+    };
+    $log->logdie("$@") if $@;
+    eval {
+        $key2 = parse_neb_key($key2);
+    };
+    $log->logdie("$@") if $@;
 
     my $dbidx1 = $self->_db_index_for_key($key1);
     my $dbidx2 = $self->_db_index_for_key($key2);
-    die "cannot swap keys not stored on the same database" unless ($dbidx1 == $dbidx2);
+    $log->logdie("cannot swap keys not stored on the same database")
+        unless ($dbidx1 == $dbidx2);
 
     my $dbh1 = $self->_db_for_index($dbidx1);
     my $dbh2 = $self->_db_for_index($dbidx2);
-    die "different db handles for the same db?" unless ($dbh1 == $dbh2);
-
-    $log->debug("entered - @_");
+    $log->logdie("different db handles for the same db?")
+        unless ($dbh1 == $dbh2);
 
     # order of operations for the swap with a single db is:
@@ -543 +623 @@
               unless ($rows == 1) {
                   $query->finish;
-                  $log->logdie("affected row count is $rows instead of 1");
+                  die("affected row count is $rows instead of 1");
               }
           }
@@ -556 +636 @@
               unless ($rows == 1) {
                   $query->finish;
-                  $log->logdie("affected row count is $rows instead of 1");
+                  die("affected row count is $rows instead of 1");
               }
           }
@@ -569 +649 @@
               unless ($rows == 1) {
                   $query->finish;
-                  $log->logdie("affected row count is $rows instead of 1");
+                  die("affected row count is $rows instead of 1");
               }
           }
@@ -579 +659 @@
             $db->rollback;
             $log->debug("rollback");
-            if ($@ =~ /Deadlock found/) {
-                $log->warn("database deadlock retrying transaction: $@");
+            if ($@ =~ $trans_regex) {
+                $log->warn("database error, retrying transaction: $@");
+                sleep TRANS_RETRY_WAIT;
                 redo TRANS;
             }
@@ -612 +693 @@
     my $self = shift;
 
+    my $log = $self->log;
+    $log->debug("entered - @_");
+
     my ($key, $vol_name) = validate_pos(@_,
         {
@@ -633 +717 @@
     );
 
+    my $sql = $self->sql;
+
     # if a volume name is explicity specified then we should make the
     # replication onto that volume (even if there is alread an instance on that
@@ -642 +728 @@
 
     # vol_name overrides the key implied volume
-    $key = parse_neb_key($key, $vol_name);
+    eval {
+        $key = parse_neb_key($key, $vol_name);
+    };
+    $log->logdie("$@") if $@;
     $vol_name = $key->volume;
 
-    my $log = $self->log;
-    my $sql = $self->sql;
     my $db  = $self->db($key);
-
-    $log->debug("entered - @_");
 
     if (defined $vol_name
         and not $self->_is_valid_volume_name($key, $key->volume)) {
-        if ($key->soft_volume) {
+        unless ($key->hard_volume) {
             $log->warn( "$vol_name is not a known volume name" );
             $vol_name = undef;
         } else {
-            die "$vol_name is not a valid volume name"
+           $log->logdie("$vol_name is not a valid volume name");
         }
     }
@@ -681 +766 @@
                 unless ( $rows > 0 ) {
                     $query->finish;
-                    $log->logdie( "storage object does not exist" );
+                    die( "storage object does not exist" );
                 }
 
@@ -725 +810 @@
             $db->rollback;
             # handle soft volumes
-            if (defined $vol_name and defined $key->soft_volume) {
+            if (defined $vol_name and not defined $key->hard_volume) {
                 $log->debug("retrying with 'any' volume");
                 return $self->replicate_object($key->path, 'any');
             }
             $log->debug("rollback");
-            if ($@ =~ /Deadlock found/) {
-                $log->warn("database deadlock retrying transaction: $@");
+            if ($@ =~ $trans_regex) {
+                $log->warn("database error, retrying transaction: $@");
+                sleep TRANS_RETRY_WAIT;
                 redo TRANS;
             }
@@ -739 +825 @@
     }
 
+    # check to see if the user.mode xattr exists
+    eval {
+        my $mode = $self->getxattr_object("$key", 'user.mode');
+        if (defined $mode) {
+            $self->chmod_object("$key", $mode);
+        }
+    };
+    if ($@) {
+        unless ($@ =~ qr/user.mode does not exist/) {
+            $log->logdie("error: $@");
+        }
+    }
+
     $log->debug("leaving");
 
@@ -748 +847 @@
 {
     my $self = shift;
+
+    my $log = $self->log;
+    $log->debug( "entered - @_" );
 
     my ( $key, $type ) = validate_pos( @_,
@@ -764 +866 @@
     );
 
+    my $sql = $self->sql;
+
     # ignore volume
-    $key = parse_neb_key($key);
-
-    my $log = $self->log;
-    my $sql = $self->sql;
+    eval {
+        $key = parse_neb_key($key);
+    };
+    $log->logdie("$@") if $@;
+
     my $db  = $self->db($key);
-
-    $log->debug( "entered - @_" );
 
     my $so_id;
@@ -785 +888 @@
                 unless ( $rows == 1 ) {
                     $query->finish;
-                    $log->logdie( "storage object does not exist" );
+                    die( "storage object does not exist" );
                 }
 
@@ -800 +903 @@
                 # can't set a write lock if there are read locks
                 if ($write_lock) {
-                    $log->logdie("can not write lock twice -- retry");
+                    die("can not write lock twice -- retry");
                 }
                 
                 if ($read_lock > 0) {
-                    $log->logdie("can not write lock after read lock -- retry");
+                    die("can not write lock after read lock -- retry");
                 }
 
@@ -813 +916 @@
                     # 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");
+                        die("affected row count is $rows instead of 1");
                     }
 
@@ -820 +923 @@
                 # 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");
+                    die("can not read lock after write lock -- retry");
                 }
 
@@ -829 +932 @@
                     # 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");
+                        die("affected row count is $rows instead of 1");
                     }
                 }
@@ -840 +943 @@
             $db->rollback;
             $log->debug("rollback");
-            if ($@ =~ /Deadlock found/) {
-                $log->warn("database deadlock retrying transaction: $@");
+            if ($@ =~ $trans_regex) {
+                $log->warn("database error, retrying transaction: $@");
+                sleep TRANS_RETRY_WAIT;
                 redo TRANS;
             }
@@ -858 +962 @@
 {
     my $self = shift;
+
+    my $log = $self->log;
+    $log->debug( "entered - @_" );
 
     my ( $key, $type ) = validate_pos( @_,
@@ -874 +981 @@
     );
 
+    my $sql = $self->sql;
+
     # ignore volume
-    $key = parse_neb_key($key);
-
-    my $log = $self->log;
-    my $sql = $self->sql;
+    eval {
+        $key = parse_neb_key($key);
+    };
+    $log->logdie("$@") if $@;
+
     my $db  = $self->db($key);
-
-    $log->debug( "entered - @_" );
 
     my $so_id;
@@ -895 +1003 @@
                 unless ($rows == 1) {
                     $query->finish;
-                    $log->logdie("storage object does not exist");
+                    die("storage object does not exist");
                 }
 
@@ -909 +1017 @@
                 # 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");
+                    die("can not have a write lock under a read lock");
                 }
 
                 unless ($write_lock) {
-                    $log->logdie("can not remove non-existant write lock");
+                    die("can not remove non-existant write lock");
                 }
 
@@ -922 +1030 @@
                     # 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");
+                        die("affected row count is $rows instead of 1");
                     }
                 }
@@ -929 +1037 @@
                 # 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");
+                    die("can not have a read lock under a write lock");
                 }
                    
                 if ($read_lock == 0) {
-                    $log->logdie("can not remove non-existant read lock");
+                    die("can not remove non-existant read lock");
                 }
 
@@ -942 +1050 @@
                     # 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");
+                        die("affected row count is $rows instead of 1");
                     }
 
@@ -953 +1061 @@
             $db->rollback;
             $log->debug("rollback");
-            if ($@ =~ /Deadlock found/) {
-                $log->warn("database deadlock retrying transaction: $@");
+            if ($@ =~ $trans_regex) {
+                $log->warn("database error, retrying transaction: $@");
+                sleep TRANS_RETRY_WAIT;
                 redo TRANS;
             }
@@ -971 +1080 @@
 {
     my $self = shift;
+
+    my $log = $self->log;
+    $log->debug("entered - @_");
 
     my ($key, $name, $value, $flags) = validate_pos(@_,
@@ -981 +1093 @@
         {
             type        => SCALAR,
+            callbacks   => {
+                'xattr is in user. namespace'
+                    => sub { ($_[0]) =~ qr/^user\./ },
+            },
         },
         {
@@ -993 +1109 @@
     );
 
+    my $sql = $self->sql;
+
     # ignore volume
-    $key = parse_neb_key($key);
-
-    my $log = $self->log;
-    my $sql = $self->sql;
+    eval {
+        $key = parse_neb_key($key);
+    };
+    $log->logdie("$@") if $@;
+
     my $db  = $self->db($key);
-
-    $log->debug("entered - @_");
 
 TRANS: while (1) {
@@ -1020 +1137 @@
             if ($flags eq 'create') {
                 unless ($rows == 1) {
-                    $log->logdie( "affected row count is $rows instead of 1" );
+                    die( "affected row count is $rows instead of 1" );
                 }
             } else {
@@ -1026 +1143 @@
                 # 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" );
+                    die( "affected row count is $rows instead of 2" );
                 }
             }
@@ -1036 +1153 @@
             $db->rollback;
             $log->debug("rollback");
-            if ($@ =~ /Deadlock found/) {
-                $log->warn("database deadlock retrying transaction: $@");
+            if ($@ =~ $trans_regex) {
+                $log->warn("database error, retrying transaction: $@");
+                sleep TRANS_RETRY_WAIT;
                 redo TRANS;
             }
@@ -1054 +1172 @@
 {
     my $self = shift;
+
+    my $log = $self->log;
+    $log->debug("entered - @_");
 
     my ($key, $name) = validate_pos(@_,
@@ -1064 +1185 @@
         {
             type        => SCALAR,
+            callbacks   => {
+                'xattr is in user. namespace'
+                    => sub { ($_[0]) =~ qr/^user\./ },
+            },
         },
     );
 
+    my $sql = $self->sql;
+
     # ignore volume
-    $key = parse_neb_key($key);
-
-    my $log = $self->log;
-    my $sql = $self->sql;
+    eval {
+        $key = parse_neb_key($key);
+    };
+    $log->logdie("$@") if $@;
+
     my $db  = $self->db($key);
-
-    $log->debug("entered - @_");
 
     my $value;
@@ -1085 +1211 @@
         if ($rows == 0) {
             $query->finish;
-            $log->logdie( "xattr $key:$name does not exist" );
+            die( "xattr $key:$name does not exist" );
         }
         # if we go more then one row bad something very bad has happened.
         unless ($rows == 1) {
             $query->finish;
-            $log->logdie( "affected row count is $rows instead of 1" );
+            die( "affected row count is $rows instead of 1" );
         }
 
@@ -1099 +1225 @@
         $value = $row->{ 'value' };
     };
-    $log->logdie("database error: $@") if $@;
+    if ($@) {
+        if ($@ =~ /user\..*? does not exist/) {
+            # do not log xattr does not exist messages
+            die $@;
+        }
+        $log->logdie("database error: $@") if $@;
+    }
 
     $log->debug("leaving");
@@ -1110 +1242 @@
 {
     my $self = shift;
+
+    my $log = $self->log;
+    $log->debug("entered - @_");
 
     my ($key) = validate_pos(@_,
@@ -1120 +1255 @@
     );
 
+    my $sql = $self->sql;
+
     # ignore volume
-    $key = parse_neb_key($key);
-
-    my $log = $self->log;
-    my $sql = $self->sql;
+    eval {
+        $key = parse_neb_key($key);
+    };
+    $log->logdie("$@") if $@;
+
     my $db  = $self->db($key);
-
-    $log->debug("entered - @_");
 
     my @xattrs;
@@ -1151 +1287 @@
     my $self = shift;
 
+    my $log = $self->log;
+    $log->debug("entered - @_");
+
     my ($key, $name) = validate_pos(@_,
         {
@@ -1160 +1299 @@
         {
             type        => SCALAR,
+            callbacks   => {
+                'xattr is in user. namespace'
+                    => sub { ($_[0]) =~ qr/^user\./ },
+            },
         },
     );
 
+    my $sql = $self->sql;
+
     # ignore volume
-    $key = parse_neb_key($key);
-
-    my $log = $self->log;
-    my $sql = $self->sql;
+    eval {
+        $key = parse_neb_key($key);
+    };
+    $log->logdie("$@") if $@;
+
     my $db  = $self->db($key);
-
-    $log->debug("entered - @_");
 
 TRANS: while (1) {
@@ -1179 +1323 @@
             $query->finish;
 
+            # no rows affected means the xattr did not exist
+            if ($rows == 0) {
+                die( "xattr $key:$name does not exist" );
+            }
+
             # 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" );
+            if ($rows > 1) {
+                die( "affected row count is $rows instead of 1" );
             }
 
@@ -1190 +1339 @@
             $db->rollback;
             $log->debug("rollback");
-            if ($@ =~ /Deadlock found/) {
-                $log->warn("database deadlock retrying transaction: $@");
+            if ($@ =~ $trans_regex) {
+                $log->warn("database error, retrying transaction: $@");
+                sleep TRANS_RETRY_WAIT;
                 redo TRANS;
             }
@@ -1209 +1359 @@
     my $self = shift;
 
+    my $log = $self->log;
+    $log->debug( "entered - @_" );
+
     my ($pattern) = validate_pos( @_,
         {
@@ -1216 +1369 @@
     );
 
-    $pattern = parse_neb_key($pattern);
-
-    my $log = $self->log;
-
-    $log->debug( "entered - @_" );
+
+    eval {
+        $pattern = parse_neb_key($pattern) if defined $pattern;
+    };
+    $log->logdie("$@") if $@;
 
     unless (defined $pattern) {
@@ -1235 +1388 @@
     $log->logdie("no keys found") unless ( scalar @keys );
 
+    if (defined $self->cache) {
+        foreach my $path (@keys) {
+            $self->cache->set($path, 1);
+                $log->debug("key added to cache as: $path");
+        }
+    }
+
     $log->debug( "leaving" );
 
@@ -1245 +1405 @@
 
     my $self    = shift;
+
+    my $log = $self->log;
+    $log->debug( "entered - @_" );
+
     my $index   = shift;
     my $key     = shift;
 
-    my $log = $self->log;
     my $sql = $self->sql;
     my $db  = $self->_db_for_index($index);
 
-    $log->debug( "entered - @_" );
 
     # first check to see if the key is an exact match
@@ -1310 +1472 @@
     my $self = shift;
 
+    my $log = $self->log;
+    $log->debug("entered - @_");
+
     my ($key, $vol_name) = validate_pos(@_,
         {
@@ -1330 +1495 @@
     );
 
+    my $sql = $self->sql;
+
+#    unless ($key) {
+#        $log->warn("key was undefined after validate_pos(), trying again...");
+#        return $self->find_instances(@_);
+#    }
+
     # vol_name overrides the key implied volume
-    $key = parse_neb_key($key, $vol_name);
+    eval {
+        $key = parse_neb_key($key, $vol_name);
+    };
+    $log->logdie("$@") if $@;
     $vol_name = $key->volume;
 
-    my $log = $self->log;
-    my $sql = $self->sql;
     my $db  = $self->db($key);
-
-    $log->debug("entered - @_");
 
     # the key's volume can't be validiated on input for this method so we have
@@ -1344 +1515 @@
     if (defined $vol_name
         and not $self->_is_valid_volume_name($key, $key->volume)) {
-        if ($key->soft_volume) {
+        if ($key->hard_volume) {
+            $log->logdie("$vol_name is not a valid volume name");
+        } else {
             $log->warn( "$vol_name is not a known volume name" );
             $vol_name = undef;
-        } else {
-            die "$vol_name is not a valid volume name"
         }
     }
@@ -1361 +1532 @@
             unless ($rows > 0) {
                 $query->finish;
-                $log->logdie("no instances on storage volume or volume is not avaiable for key: $key volume: $vol_name");
+                die("no instances on storage volume or volume is not avaiable for key: $key volume: $vol_name");
             }
         } else {
@@ -1369 +1540 @@
             unless ($rows > 0) {
                 $query->finish;
-                $log->logdie("no instances available for key: $key");
+                die("no instances available for key: $key");
             }
         }
@@ -1381 +1552 @@
         $db->rollback;
         # handle soft volumes
-        if (defined $vol_name and defined $key->soft_volume) {
+        if (defined $vol_name and not defined $key->hard_volume) {
             $log->debug("retrying with 'any' volume");
             return $self->find_instances($key->path, 'any');
@@ -1403 +1574 @@
     my $self = shift;
 
+    my $log = $self->log;
+    $log->debug( "entered - @_" );
+
     my ($key, $uri) = validate_pos( @_,
         {
@@ -1415 +1589 @@
     );
 
+    my $sql = $self->sql;
+
     # ignore volume
-    $key = parse_neb_key($key);
-
-    my $log = $self->log;
-    my $sql = $self->sql;
+    eval {
+        $key = parse_neb_key($key);
+    };
+    $log->logdie("$@") if $@;
+
     my $db  = $self->db($key);
-
-    $log->debug( "entered - @_" );
 
 TRANS: while (1) {
         eval {
+            my $instances;
             my $so_id;
-            my $instances;
-            # get so_id
+            my $ins_id;
+            # find so_id for key and get count of instances
             {
-                my $query = $db->prepare_cached( $sql->get_object_from_uri );
-                my $rows = $query->execute( $uri );
-
+                my $query = $db->prepare_cached( $sql->get_instance_count_by_ext_id );
+                my $rows = $query->execute($key->path);
                 unless ( $rows > 0 ) {
                     $query->finish;
-                    $log->logdie( "no instance is associated with uri" );
-                }
-
-                $so_id = $query->fetchrow_hashref->{ 'so_id' };
+                    die( "$key has no associated instances - this should not happen" );
+                }
+
+                my $record = $query->fetchrow_hashref;
+                $so_id      = $record->{ 'so_id' };
+                $instances  = $record->{ 'count(ins_id)' };
                 $query->finish;
-
-            }
-
+            }
+
+            # find ins_id for uri
             {
-                my $query = $db->prepare_cached( $sql->get_instance_count );
-                $query->execute( $so_id );
-
-                $instances = $query->fetchrow_hashref->{ 'count(ins_id)' };
+                my $query = $db->prepare_cached( $sql->get_instance_by_uri );
+                my $rows = $query->execute($so_id, $uri);
+                unless ( $rows > 0) {
+                    $query->finish;
+                    die( "no instance is associated with uri" );
+                }
+
+                $ins_id = $query->fetchrow_hashref->{ 'ins_id' };
                 $query->finish;
             }
@@ -1453 +1634 @@
             # remove instance
             {
-                my $query = $db->prepare_cached( $sql->delete_instance );
-                my $rows = $query->execute( $uri );
+                my $query = $db->prepare_cached( $sql->delete_instance_by_ins_id );
+                my $rows = $query->execute( $ins_id );
                 $query->finish;
                 
-                # if we affected something other then two rows something very bad
-                # has happened
+                # if we affected something other then one row something very
+                # bad has happened
                 unless ( $rows == 1 ) {
-                    $log->logdie( "affected row count is $rows instead of 1" );
+                    die( "affected row count is $rows instead of 1" );
                 }
             }
@@ -1467 +1648 @@
             # remove it too
             if ( $instances == 1 ) {
+                # remove key from cache
+                $self->cache->delete($key->path) if defined $self->cache;
+
                 # we just removed the last instance
                 my $query = $db->prepare_cached( $sql->delete_object );
@@ -1474 +1658 @@
                 # 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" );
+                    die( "affected row count is $rows instead of 2" );
                 }
             }
@@ -1484 +1668 @@
             $db->rollback;
             $log->debug("rollback");
-            if ($@ =~ /Deadlock found/) {
-                $log->warn("database deadlock retrying transaction: $@");
+            if ($@ =~ $trans_regex) {
+                $log->warn("database error, retrying transaction: $@");
+                sleep TRANS_RETRY_WAIT;
                 redo TRANS;
             }
@@ -1502 +1687 @@
 {
     my $self = shift;
+
+    my $log = $self->log;
+    $log->debug("entered - @_");
 
     my ( $key ) = validate_pos( @_,
@@ -1512 +1700 @@
     );
 
+    my $sql = $self->sql;
+
     # ignore volume
-    $key = parse_neb_key($key);
-
-    my $log = $self->log;
-    my $sql = $self->sql;
+    eval {
+        $key = parse_neb_key($key);
+    };
+    $log->logdie("$@") if $@;
+
     my $db  = $self->db($key);
-
-    $log->debug("entered - @_");
 
     my $stat;
@@ -1527 +1716 @@
 
         unless ($rows == 1) {
-            $log->logdie("no storage object found");
+            die("no storage object found");
         }
 
@@ -1548 +1737 @@
     my $self = shift;
 
+    my $log = $self->log;
+    $log->debug("entered - @_");
+
     validate_pos(@_); 
 
-    my $log = $self->log;
     my $sql = $self->sql;
     my $db  = $self->_db_for_index(0); # XXX fix as above
 
-    $log->debug("entered - @_");
-
     my $stats;
     my $query;
@@ -1576 +1765 @@
 }
 
+sub chmod_object
+{
+    my $self = shift;
+
+    my $log = $self->log;
+    $log->debug("entered - @_");
+
+    my ($key, $mode) = validate_pos( @_, 
+        {
+            type        => SCALAR,
+            callbacks   => {
+                'is valid object key' => sub { $self->_is_valid_object_key($_[0]) },
+            },
+        },
+        {
+            type        => SCALAR,
+            regex       => qr/\d{3,4}/,
+            callbacks   => {
+                'is allowable mode' => sub {
+                    $_[0] == (S_IRUSR | S_IRGRP)
+                },
+            },
+        },
+    );
+
+    my $sql = $self->sql;
+
+    # ignore volume
+    eval {
+        $key = parse_neb_key($key);
+    };
+    $log->logdie("$@") if $@;
+
+    my $db  = $self->db($key);
+
+    # find all instances of this object
+    my $locations;
+    eval {
+        $locations = $self->find_instances("$key");
+    };
+    $log->logdie("error: $@") if $@;
+
+    # update each instances
+    foreach my $inst (@$locations) {
+        my $path = URI->new($inst)->path;
+
+        $self->_retry(sub { chmod($mode, $path) })
+            or $log->logdie("can not chmod() $path: $!");
+
+        # XXX I'm assuming that it's OK to fsync() a filehandle that's only
+        # open for reading?  Opening as w/rw here can fail if the chmod removes
+        # write permissions.
+        my $fh;
+        $self->_retry(sub { open($fh, '<', $path) })
+            or $log->logdie("can not open() $path: $!");
+
+        # fsync(3c)
+        $self->_retry(sub { $fh->sync() })
+            or $log->logdie("can not sync() $path: $!");
+
+        $self->_retry(sub { close($fh) })
+            or $log->logdie("can not close() $path: $!");
+    }
+
+    # stick an xattr on this object with the mode
+    # XXX this would probably be better as a field in the storage_object_attr
+    # table but since we're not planning to use this for very many objects (as
+    # a %) it may not be worth adding the extra field at this time.
+    eval {
+        $self->setxattr_object("$key", 'user.mode', $mode, 'replace');
+    };
+    $log->logdie("error: $@") if $@;
+
+    $log->debug("leaving");
+
+    return $mode;
+}
 
 sub _get_storage_volume
@@ -1581 +1847 @@
     my $self = shift;
 
-    my ($key, $name, $soft_volume) = @_;
-
-    my $log = $self->log;
-    my $sql = $self->sql;
-    my $db  = $self->db($key);
-
+    my $log = $self->log;
     no warnings qw( uninitialized );
     $log->debug( "entered - @_" );
     use warnings;
+
+    my ($key, $name, $hard_volume) = @_;
+
+    my $sql = $self->sql;
+    my $db  = $self->db($key);
 
     my ($vol_id, $vol_host, $vol_path, $xattr);
@@ -1605 +1871 @@
                 # if a volume name was specified, and is soft, and we failed to
                 # find it, fall back to any volume
-                if ($soft_volume) {
+                unless ($hard_volume) {
                     ($vol_id, $vol_host, $vol_path, $xattr) = $self->_get_storage_volume($key);
                     return; # this just returns out of the eval not from the subroutine
                 }
-                $log->logdie("storage volume: $name is not available");
+                die("storage volume: $name is not available");
             }
             # when matching by name we shouldn't ever match more than once
             if ($rows > 1) {
                 $query->finish;
-                $log->logdie("affected row count is $rows instead of 1");
+                die("affected row count is $rows instead of 1");
             }
         } else {
@@ -1623 +1889 @@
             unless ($rows > 0) {
                 $query->finish;
-                $log->logdie("no storage volume is available");
+                die("no storage volume is available for key: $key volume: $name hard_volume: $hard_volume");
             }
         }
@@ -1631 +1897 @@
         $query->finish;
     };
-    $log->logdie("database error: $@") if $@;
+    if ($@) {
+        if ($@ =~ qr/no storage volume is available/) {
+            # this should not happen unless all volumes are full
+            $log->error($@);
+            $log->debug("retrying...");
+            return $self->_get_storage_volume(@_);
+        } 
+        # else
+        $log->logdie("database error: $@");
+    }
 
     $log->logdie("failed to find a suitable volume" )
@@ -1646 +1921 @@
     my $self = shift;
 
-    my $key = shift;
-
-    my $log = $self->log;
-    my $sql = $self->sql;
-    my $db  = $self->db($key);
-
+    my $log = $self->log;
     no warnings qw( uninitialized );
     $log->debug( "entered - @_" );
     use warnings;
+
+    my $key = shift;
+
+    my $sql = $self->sql;
+    my $db  = $self->db($key);
 
     my ($vol_id, $vol_host, $vol_path, $xattr);
@@ -1665 +1940 @@
         unless ($rows > 0) {
             $query->finish;
-            $log->logdie("can't find a suitable storage volume to replicate $key to");
+            die("can't find a suitable storage volume to replicate $key to");
         }
         # when matching by name we shouldn't ever match more than once
         if ($rows > 1) {
             $query->finish;
-            $log->logdie("affected row count is $rows instead of 1");
+            die("affected row count is $rows instead of 1");
         }
 
@@ -1690 +1965 @@
 sub _is_valid_object_key
 {
-    my ($self, $key) = @_;
-
-    $key = parse_neb_key($key);
-
-    my $log = $self->log;
+    my $self = shift;
+
+    my $log = $self->log;
+    $log->debug( "entered - @_" );
+
+    my ($key) = @_;
+
     my $sql = $self->sql;
+
+    eval {
+        $key = parse_neb_key($key);
+    };
+    $log->logdie("$@") if $@;
+
+    # check cache first
+    my $cached = $self->cache->get($key->path) if defined $self->cache;
+    if (defined $cached) {
+        $log->debug( "key $key found in cache as ", $key->path );
+        $log->debug( "leaving" );
+        return 1;
+    } else {
+        $log->debug( "key $key not found in cache" );
+    }
+
     my $db  = $self->db($key);
 
@@ -1712 +2005 @@
 
     if (defined $ext_id) {
+        $log->debug( "key found in db" );
+        # add key to cache
+        $self->cache->set($key->path, 1) if defined $self->cache;
+        $log->debug( "key added to cache as ", $key->path );
+        $log->debug( "leaving" );
         return 1;
     } 
 
+    $log->debug( "key not found in db" );
+    $log->debug( "leaving" );
+
     return;
 }
@@ -1721 +2022 @@
 sub _is_valid_volume_name
 {
-    my ($self, $key, $vol_name) = @_;
-
-    $key = parse_neb_key($key);
+    my $self = shift;
+
+    my $log = $self->log;
+    $log->debug( "entered - @_" );
+
+    my ($key, $vol_name) = @_;
+
+    my $sql = $self->sql;
+
+    # the volume name implied by the key is ignored.  $key is only needed to
+    # select a database connection
+    eval {
+        $key = parse_neb_key($key);
+    };
+    $log->logdie("$@") if $@;
     my $volume_info = parse_neb_volume($vol_name);
 
-    my $log = $self->log;
-    my $sql = $self->sql;
     my $db  = $self->db($key);
 
@@ -1734 +2045 @@
     # handle "any" volume
     if ($vol_name eq 'any') {
+        $log->debug( "found volume name $vol_name" );
+        $log->debug( "leaving" );
         return 1;
     }
@@ -1748 +2061 @@
 
     if (defined $vol_id and defined $vol_path) {
+        $log->debug( "found volume name $vol_name" );
+        $log->debug( "leaving" );
         return 1;
     } 
 
+    $log->debug( "volume name $vol_name not found" );
+    $log->debug( "leaving" );
+
     return;
 }
@@ -1759 +2077 @@
     my $self = shift;
 
+    my $log = $self->log;
+    $log->debug( "entered - @_" );
+
     my ($key, $so_id, $ins_id, $vol_path, $xattr) =  @_;
 
-    my $log = $self->log;
     my $sql = $self->sql;
     my $db  = $self->db($key);
@@ -1770 +2090 @@
         my $storage_filename = $self->_generate_storage_filename($key->path, $ins_id);
         unless (-d $storage_path) {
-            _retry(sub { mkpath(@_) }, $storage_path, 0, 0775)
-                or die "can't create storage path: $storage_path";
+            $self->_retry(sub { mkpath([$storage_path], 0, 0775) })
+                or die("can't create storage path: $storage_path");
         }
         # check to make sure at least the parent directory has the proper
         # permissions
-        my $mode = [_retry(sub { stat($storage_path) } )]->[2] & 07777;
+        my $mode = [$self->_retry(sub { stat($storage_path) } )]->[2] & 07777;
         unless ($mode == 0775) {
-            $log->error("$storage_path has the wrong permissions of: %04x", $mode);
-            _retry(sub { chmod(0775, $storage_path) }) or die "can't chmod $storage_path: $!";
+            # XXX: this problem is so common that it's flooding the logs
+            $log->debug("$storage_path has the wrong permissions of: 0", sprintf("%o", $mode));
+            $self->_retry(sub { chmod(0775, $storage_path) })
+                or die("can not chmod() $storage_path: $!");
         }
 
@@ -1787 +2109 @@
     };
     if ($@) {
+        if (defined $uri and -e $uri->file) {
+            unlink($uri->file)
+                or $log->error("failed to unlink() $uri: $!");
+        }
         $log->logdie($@);
     }
@@ -1792 +2118 @@
     if ($xattr) {
         my $path = $uri->file;
-        die "can not set xattr on $path: $!"
+        $log->logdie("can not set xattr on $path: $!")
             unless (setfattr($path, 'user.nebulous_key', $key->path));
     }
@@ -1804 +2130 @@
     my $self = shift;
 
+    my $log = $self->log;
+    $log->debug( "entered - @_" );
+
     my ($path) = @_;
 
     # perl's open() can't do an O_CREAT | O_EXCL
-    die "file $path already exists" if (-e $path);
+    # XXX is it possible to tell if this system call failed?
+    -e $path
+        and $log->logdie("file $path already exists");
 
     my $fh;
-    die "can not open $path: $!"
-        unless (_retry(sub { open($fh, '>', $path) }));
+    $self->_retry(sub { open($fh, '>', $path) })
+        or $log->logdie("can not open() $path: $!");
 
     # chmod before fsync() to make sure the changed perms hit the disk too
-    die "can not chmod $path: $!"
-        unless (_retry(sub { chmod(0664, $path) }));
-
-    die "can not flush $path: $!"
-        unless ($fh->flush);
-
-    die "can not sync $path: $!"
-        unless ($fh->sync);
-
-    die "can not close $path: $!"
-        unless (_retry(sub { close($fh) }));
+    $self->_retry(sub { chmod(0664, $path) })
+        or $log->logdie("can not chmod() $path: $!");
+
+    $self->_retry(sub { $fh->sync() })
+        or $log->logdie("can not sync() $path: $!");
+
+    $self->_retry(sub { close($fh) })
+        or $log->logdie("can not close() $path: $!");
 
     return $path;
@@ -1861 +2189 @@
 sub _retry
 {
+    my $self = shift;
+
+    my $log = $self->log;
+    $log->debug( "entered - @_" );
+
     my $func = shift;
 
@@ -1869 +2202 @@
         };
         if ($@) {
-            die $@;
-            sleep 1;
+            $log->logdie($@);
+            sleep NFS_RETRY_WAIT;
             next;
         }
@@ -1879 +2212 @@
     # if the loop ended and $@ is set, rethrow the error
     if ($@) {
-        die $@;
+        $log->logdie($@);
     }