Index: trunk/Nebulous-Server/lib/Nebulous/Server.pm
===================================================================
--- trunk/Nebulous-Server/lib/Nebulous/Server.pm	(revision 23698)
+++ trunk/Nebulous-Server/lib/Nebulous/Server.pm	(revision 23932)
@@ -9,5 +9,5 @@
 no warnings qw( uninitialized );
 
-our $VERSION = '0.15';
+our $VERSION = '0.17';
 
 use base qw( Class::Accessor::Fast );
@@ -15,14 +15,14 @@
 use DBI;
 use Digest::SHA1 qw( sha1_hex );
-use File::Basename qw( dirname );
+use File::Basename qw( basename dirname fileparse );
 use File::ExtAttr qw( setfattr );
 use File::Path;
 use File::Spec;
-use Log::Log4perl;
-use Nebulous::Keys qw( parse_neb_key parse_neb_volume );
+use Log::Log4perl qw( :levels );
+use Nebulous::Key qw( parse_neb_key parse_neb_volume );
 use Nebulous::Server::Config;
 use Nebulous::Server::Log;
 use Nebulous::Server::SQL;
-use Params::Validate qw( validate_pos SCALAR SCALARREF UNDEF );
+use Params::Validate qw( validate validate_pos SCALAR SCALARREF UNDEF BOOLEAN );
 use URI::file;
 
@@ -37,9 +37,18 @@
 
     # let Nebulous::Server::Config validate our params
-    my $config = Nebulous::Server::Config->init( @_ );
+    my $config = Nebulous::Server::Config->new( @_ );
+
+    return $class->new_from_config($config);
+}
+
+
+sub new_from_config
+{
+    my ($class, $config) = @_;
 
     # log4perl is not avaliable until we call init()
     Nebulous::Server::Log->init($config);
     my $log = Log::Log4perl::get_logger( "Nebulous::Server" );
+    $log->level($config->trace);
 
     my $sql = Nebulous::Server::SQL->new;
@@ -52,21 +61,34 @@
     $self->config($config);
 
-    # ask for the db handle as a means of validating the database parameters
-    $self->db;
-
     $log->debug( "leaving" );
-    
+
     return $self;
 }
 
-
-sub db
-{
-    my $self = shift;
-
-    if (@_) {
-        $self->{db} = $_[0];
-        return $self;
-    }
+# EAM : In the 'distributed' version of Nebulous, there is a collection of N databases
+# the db_index uniquely defines the db used by a given key
+sub _db_index_for_key
+{
+    my ($self, $key) = @_;
+
+    my $config  = $self->config;
+
+    my $db_index = 0;
+    die "key not defined" unless defined $key;
+
+    # hash the key to select the correct database instance
+    # only use the first 8 hex chars... have to be careful to avoid an int
+    # overflow here
+
+    # hash only the directory component of the path and not the filename
+    my $path = dirname($key->path);
+    $db_index = unpack("h8", sha1_hex($path)) % $config->n_db;
+
+    return $db_index;
+}
+
+sub _db_for_index
+{
+    my ($self, $db_index) = @_;
 
     my $log     = $self->log;
@@ -74,11 +96,18 @@
     my $config  = $self->config;
 
+    # lookup to see if we have a stored dbh for this database
+    my $dbh = $self->{dbs}[$db_index];
     # if the dbh is still alive, return it
-    if (defined $self->{db} and $self->{db}->ping) {
+    if (defined $dbh and $dbh->ping) {
         $log->debug("db handle is still alive");
-        return $self->{db};
+        return $dbh;
     }
     # otherwise create a new connection
     $log->debug("db handle is dead/unopened");
+
+    # lookup database info
+    my $db_config = $config->db($db_index);
+    die "can't find database configuration info for database # $db_index"
+        unless $db_config;
 
     # if we're running under mod_perl & Apache::DBI is loaded we want to
@@ -87,10 +116,9 @@
     # processes and the database might have gone away on us.  Apache::DBI will
     # take care of getting a valid dbh back.
-    my $db;
     eval {
-        $db = DBI->connect_cached(
-            $config->dsn,
-            $config->dbuser,
-            $config->dbpasswd,
+        $dbh = DBI->connect_cached(
+            $db_config->dsn,
+            $db_config->dbuser,
+            $db_config->dbpasswd,
             {
                 RaiseError => 1,
@@ -100,20 +128,40 @@
         );
 
-        $db->do( $sql->set_transaction_model );
-        $log->debug( "connected to database: ", sub { $db->data_sources; } );
-        $db->commit;
+        $dbh->do( $sql->set_transaction_model );
+        $log->debug( "connected to database: ", sub { $dbh->data_sources; } );
+        $dbh->commit;
         $log->debug("commit");
     };
     if ( $@ ) {
-        $db->rollback if $db;
+        $dbh->rollback if $dbh;
         $log->debug("rollback");
         $log->logdie( "database error: $@" );
     }
 
-    $self->{db} = $db;
-
-    return $db;
-}
-
+    $self->{dbs}[$db_index] = $dbh;
+
+    return $dbh;
+}
+
+sub db
+{
+    my $self = shift;
+
+    my ($key) = validate_pos(@_,
+        {
+            isa => 'Nebulous::Key',
+        },
+    );
+
+    my $log     = $self->log;
+    my $sql     = $self->sql;
+    my $config  = $self->config;
+
+    die "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;
+}
 
 sub create_object
@@ -138,18 +186,18 @@
     );
 
-    my $log = $self->log;
-    my $sql = $self->sql;
-    my $db  =$self->db;
-
-    $log->debug( "entered - @_" );
-
     # vol_name overrides the key implied volume
     $key = parse_neb_key($key, $vol_name);
     $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
     # to check it after parsing the key
     if (defined $vol_name
-        and not $self->_is_valid_volume_name($key->volume)) {
+        and not $self->_is_valid_volume_name($key, $key->volume)) {
         if ($key->soft_volume) {
             $log->warn( "$vol_name is not a known volume name" );
@@ -161,5 +209,7 @@
         
     my ($vol_id, $vol_host, $vol_path, $vol_xattr)
-        = $self->_get_storage_volume($vol_name, $key->soft_volume);
+        = $self->_get_storage_volume($key, $vol_name, $key->soft_volume);
+
+    my $parent_id = $self->_resolve_dir_parent_id(key => $key, create => 1);
 
     my $uri;
@@ -169,5 +219,5 @@
                 # create storage_object
                 my $query = $db->prepare_cached( $sql->new_object ); 
-                $query->execute('NULL', $key->path);
+                $query->execute('NULL', $key->path, basename($key->path), $parent_id);
             }
 
@@ -215,5 +265,5 @@
 
             # 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);
+            $uri = $self->_create_empty_instance_file($key, $so_id, $ins_id, $vol_path, $vol_xattr);
             $log->debug("created $uri on volume ID: $vol_id");
 
@@ -247,4 +297,125 @@
 
 
+sub _resolve_dir_parent_id
+{
+    my $self = shift;
+
+    my %p = validate(@_,
+        {
+            key     => {
+                isa         => 'Nebulous::Key',
+            },
+            create  => {
+                type        => BOOLEAN,
+                optional    => 1,
+                default     => undef,
+            },
+            # return dir_id instead of parent_id
+            dir_id  => {
+                type        => BOOLEAN,
+                optional    => 1,
+                default     => undef,
+            },
+        }
+    );
+
+    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
+    return 1 if $key->path eq '';
+
+    # resolve parent directory
+    my @dirs;
+
+    # File::Spec->splitpath was causing ->splitdir to always an extra dir
+    # named "" because of a trailing /
+    unless ($p{dir_id}) {
+        @dirs = File::Spec->splitdir(dirname($key->path));
+    } else {
+        @dirs = File::Spec->splitdir($key->path);
+    }
+    # dirname returns "." if there is no dir component to the path, we have
+    # to filter this out
+    @dirs = grep(!/^\.$/, @dirs);
+
+    $log->debug("looking for dirs - ", join(" : ", @dirs), "\n");
+
+    # start at the root dir; '/' == 1
+    my $parent_id = 1;
+    my $dir_id;
+TRANS: while (1) {
+        eval {
+            foreach my $dir (@dirs) {
+                $dir_id = undef;
+                {
+                    my $query = $db->prepare_cached($sql->get_directory); 
+                    $query->execute($parent_id, $dir);
+                    if ($query->rows) {
+                        $dir_id = $query->fetchrow_hashref->{'dir_id'};
+                        $log->debug("resolved $dir to dir_id: $dir_id");
+                    }
+                    $query->finish;
+                }
+
+                # if we found a dir_id, a row for this directory already exists
+                if (defined $dir_id) {
+                    $parent_id = $dir_id;
+                    # note that you can't exit an eval {} with next
+                    next;
+                }
+
+                # else dir doesn't exist
+                unless ($p{create}) {
+                    # resolution failed
+                    $parent_id = undef;
+                    last;
+                }
+
+                {
+                    # dir doesn't exist, create it
+                    my $query = $db->prepare_cached($sql->new_directory);
+                    $query->execute($dir, $parent_id);
+                }
+
+                # get the dir_id of the new directory entry 
+                {
+                    my $query = $db->prepare_cached($sql->last_insert_id);
+                    $query->execute();
+
+                    # the new dir_id will be the parent_id of the next
+                    # descendent directory
+                    ($parent_id) = $query->fetchrow_array;
+                    $query->finish;
+                }
+                $log->logdie("failed to get LAST_INSERT_ID()")
+                    unless $parent_id;
+
+                $db->commit;
+            }
+        };
+        if ($@) {
+            $db->rollback;
+            $log->debug("rollback");
+            if ($@ =~ /Deadlock found/) {
+                $log->warn("database deadlock retrying transaction: $@");
+                redo TRANS;
+            }
+            $log->logdie("error: $@");
+        }
+        last;
+    }
+
+    $log->debug("leaving");
+
+    return $p{dir_id} ? $dir_id : $parent_id;
+}
+
+
 sub rename_object
 {
@@ -266,14 +437,20 @@
         },
     );
-
-    my $log = $self->log;
-    my $sql = $self->sql;
-    my $db  =$self->db;
-
-    $log->debug("entered - @_");
 
     # ignore volumes
     $key    = parse_neb_key($key);
     $newkey = parse_neb_key($newkey);
+
+    my $log = $self->log;
+    my $sql = $self->sql;
+    my $db  = $self->db($key);
+
+    $log->debug("entered - @_");
+
+    # XXX this may require database migration in the future
+    unless ($self->_db_index_for_key($key)
+         == $self->_db_index_for_key($newkey)) {
+        $log->logdie("can not rename objects across distributed database boundaries");
+    }
 
 TRANS: while (1) {
@@ -282,5 +459,5 @@
             my $query = $db->prepare_cached($sql->rename_object); 
             # this SQL statment takes the new key name as the first param
-            my $rows = $query->execute($newkey->path, $key->path);
+            my $rows = $query->execute($newkey->path, basename($newkey->path), $self->_resolve_dir_parent_id(key => $newkey, create => 1), $key->path);
 
             # if we affected more then one row something very bad has happened.
@@ -310,5 +487,4 @@
 }
 
-
 sub swap_objects
 {
@@ -329,10 +505,4 @@
         },
     );
-
-    my $log = $self->log;
-    my $sql = $self->sql;
-    my $db  =$self->db;
-
-    $log->debug("entered - @_");
 
     # ignore volumes
@@ -340,49 +510,66 @@
     $key2 = parse_neb_key($key2);
 
-    # order of operations for the swap:
+    my $log = $self->log;
+    my $sql = $self->sql;
+
+    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);
+
+    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 - @_");
+
+    # order of operations for the swap with a single db is:
     # key1 -> key1.swap
     # key2 -> key1
     # key1.swap -> key2
 
-TRANS: while (1) {
+    my $db = $dbh1;
+  TRANS: while (1) {
         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");
-                }
-            }
+              # key1 -> key1.swap
+              my $query = $db->prepare_cached($sql->rename_object); 
+              # this SQL statment takes the new key name as the first param
+              # XXX currently using a bogus dir_id -- this may cause a problem
+              # someday but it's unlikley as it's contained entirely in the
+              # transaction
+              my $rows = $query->execute($key1->path . ".swap", basename($key1->path) . ".swap", 1, $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, basename($key1->path), $self->_resolve_dir_parent_id(key => $key1, create => 1), $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, basename($key2->path), $self->_resolve_dir_parent_id(key => $key2, create => 1), $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;
@@ -398,6 +585,6 @@
             $log->logdie("database error: $@");
         }
-        last;
-    }
+      last;
+  }
 
     $log->debug("leaving");
@@ -406,4 +593,18 @@
 }
 
+# EAM : from JH, below is a possible option to swap instances across
+# dbs.  I recommend that this action be disallowed, and instead such
+# operations be implemented only as a combination of copy and delete
+
+# order of operations for the swap between two dbs is:
+# key1 start transaction
+# key1 -> read all instances
+# key1 -> remove all instances
+# key2 start transaction
+# key2 -> read all instances
+# key2 -> remove all instances
+# key1 -> insert key 2 instances
+# key2 -> insert key 1 instances
+# key1,2 commit
 
 sub replicate_object
@@ -440,16 +641,16 @@
     # then we should throw an error 
 
-    my $log = $self->log;
-    my $sql = $self->sql;
-    my $db  =$self->db;
-
-    $log->debug("entered - @_");
-
     # vol_name overrides the key implied volume
     $key = parse_neb_key($key, $vol_name);
     $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->volume)) {
+        and not $self->_is_valid_volume_name($key, $key->volume)) {
         if ($key->soft_volume) {
             $log->warn( "$vol_name is not a known volume name" );
@@ -463,5 +664,5 @@
     if (defined $vol_name) {
         ($vol_id, $vol_host, $vol_path, $vol_xattr)
-            = $self->_get_storage_volume($vol_name);
+            = $self->_get_storage_volume($key, $vol_name);
     } else {
         ($vol_id, $vol_host, $vol_path, $vol_xattr)
@@ -508,5 +709,7 @@
 
             # 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);
+            # XXX if this fails, it should try to generate the
+            # instance on another volume (unless !$soft_volume) 
+            $uri = $self->_create_empty_instance_file($key, $so_id, $ins_id, $vol_path, $vol_xattr);
 
             {
@@ -561,12 +764,12 @@
     );
 
-    my $log = $self->log;
-    my $sql = $self->sql;
-    my $db  =$self->db;
-
-    $log->debug( "entered - @_" );
-
     # ignore volume
     $key = parse_neb_key($key);
+
+    my $log = $self->log;
+    my $sql = $self->sql;
+    my $db  = $self->db($key);
+
+    $log->debug( "entered - @_" );
 
     my $so_id;
@@ -671,12 +874,12 @@
     );
 
-    my $log = $self->log;
-    my $sql = $self->sql;
-    my $db  =$self->db;
-
-    $log->debug( "entered - @_" );
-
     # ignore volume
     $key = parse_neb_key($key);
+
+    my $log = $self->log;
+    my $sql = $self->sql;
+    my $db  = $self->db($key);
+
+    $log->debug( "entered - @_" );
 
     my $so_id;
@@ -790,12 +993,12 @@
     );
 
-    my $log = $self->log;
-    my $sql = $self->sql;
-    my $db  =$self->db;
-
-    $log->debug("entered - @_");
-
     # ignore volume
     $key = parse_neb_key($key);
+
+    my $log = $self->log;
+    my $sql = $self->sql;
+    my $db  = $self->db($key);
+
+    $log->debug("entered - @_");
 
 TRANS: while (1) {
@@ -864,12 +1067,12 @@
     );
 
-    my $log = $self->log;
-    my $sql = $self->sql;
-    my $db  =$self->db;
-
-    $log->debug("entered - @_");
-
     # ignore volume
     $key = parse_neb_key($key);
+
+    my $log = $self->log;
+    my $sql = $self->sql;
+    my $db  = $self->db($key);
+
+    $log->debug("entered - @_");
 
     my $value;
@@ -917,12 +1120,12 @@
     );
 
-    my $log = $self->log;
-    my $sql = $self->sql;
-    my $db  =$self->db;
-
-    $log->debug("entered - @_");
-
     # ignore volume
     $key = parse_neb_key($key);
+
+    my $log = $self->log;
+    my $sql = $self->sql;
+    my $db  = $self->db($key);
+
+    $log->debug("entered - @_");
 
     my @xattrs;
@@ -960,12 +1163,12 @@
     );
 
-    my $log = $self->log;
-    my $sql = $self->sql;
-    my $db  =$self->db;
-
-    $log->debug("entered - @_");
-
     # ignore volume
     $key = parse_neb_key($key);
+
+    my $log = $self->log;
+    my $sql = $self->sql;
+    my $db  = $self->db($key);
+
+    $log->debug("entered - @_");
 
 TRANS: while (1) {
@@ -1001,10 +1204,10 @@
 }
 
-
+# loop over all db_index values, passing db_index to each call
 sub find_objects
 {
     my $self = shift;
 
-    my ( $pattern ) = validate_pos( @_,
+    my ($pattern) = validate_pos( @_,
         {
             type        => SCALAR,
@@ -1013,31 +1216,87 @@
     );
 
-    my $log = $self->log;
-    my $sql = $self->sql;
-    my $db  =$self->db;
+    $pattern = parse_neb_key($pattern);
+
+    my $log = $self->log;
 
     $log->debug( "entered - @_" );
 
-    unless ($pattern) {
+    unless (defined $pattern) {
         $log->debug( "leaving" );
         $log->logdie("no keys found");
     }
 
-    # attempt to strip off neb:// if it exists
-    $pattern =~ s|^neb:||;
-
+    my @keys = ();
+    my $n_dbs = $self->config->n_db();
+    for (my $index = 0; $index < $n_dbs; $index ++) {
+        my $newkeys = $self->_find_objects_for_index($index, $pattern);
+        push @keys, @$newkeys;
+    }
+    $log->logdie("no keys found") unless ( scalar @keys );
+
+    $log->debug( "leaving" );
+
+    return \@keys;
+}
+
+# find matching objects from the given server
+sub _find_objects_for_index
+{
+
+    my $self    = shift;
+    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
     my @keys;
     eval {
-        my $query = $db->prepare_cached( $sql->find_objects );
-        $query->execute( $pattern );
+        $log->debug("trying for an exact key match with $key");
+        my $query = $db->prepare_cached( $sql->find_object_by_ext_id );
+        $query->execute( $key->path );
+        if ($query->rows) {
+            my $ext_id = $query->fetchrow_hashref->{'ext_id'};
+            $log->debug( "pattern has an exact match" );
+            push @keys, $ext_id;
+        } else {
+            $log->debug("no exact match for key");
+        }
+        $query->finish;
+    };
+    if ($@) {
+        $db->rollback;
+        $log->logdie("database error: $@");
+    }
+
+    if (scalar @keys) {
+        # it was an exact match, so stop here
+        $log->debug("leaving");
+
+        return \@keys;
+    }
+
+    # else, assume it's a directory
+    my $dir_id = $self->_resolve_dir_parent_id(key => $key, dir_id => 1);
+    unless (defined $dir_id) {
+        $log->logdie("pattern $key does not match any key or directory");
+    }
+
+    eval {
+        $log->debug("trying for a directory match under dir: $dir_id");
+        my $query = $db->prepare_cached( $sql->find_object_by_dir_id );
+        $query->execute( $dir_id );
 
         while ( my $row = $query->fetchrow_hashref ) {
             my $key = $row->{ 'ext_id' };
             push @keys, $key if $key;
+            $log->debug( "matched $key" ) if $key;
         }
     };
     $log->logdie("database error: $@") if $@;
-
-    $log->logdie("no keys found") unless ( scalar @keys );
 
     $log->debug( "leaving" );
@@ -1071,18 +1330,18 @@
     );
 
-    my $log = $self->log;
-    my $sql = $self->sql;
-    my $db  =$self->db;
-
-    $log->debug("entered - @_");
-
     # vol_name overrides the key implied volume
     $key = parse_neb_key($key, $vol_name);
     $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
     # to check it after parsing the key
     if (defined $vol_name
-        and not $self->_is_valid_volume_name($key->volume)) {
+        and not $self->_is_valid_volume_name($key, $key->volume)) {
         if ($key->soft_volume) {
             $log->warn( "$vol_name is not a known volume name" );
@@ -1144,5 +1403,11 @@
     my $self = shift;
 
-    my ( $uri ) = validate_pos( @_,
+    my ($key, $uri) = validate_pos( @_,
+        {
+            type        => SCALAR,
+            callbacks   => {
+                'is valid object key' => sub { $self->_is_valid_object_key($_[0]) },
+            },
+        },
         {
             type => SCALAR|SCALARREF,
@@ -1150,7 +1415,10 @@
     );
 
-    my $log = $self->log;
-    my $sql = $self->sql;
-    my $db  =$self->db;
+    # ignore volume
+    $key = parse_neb_key($key);
+
+    my $log = $self->log;
+    my $sql = $self->sql;
+    my $db  = $self->db($key);
 
     $log->debug( "entered - @_" );
@@ -1244,12 +1512,12 @@
     );
 
-    my $log = $self->log;
-    my $sql = $self->sql;
-    my $db  =$self->db;
-
-    $log->debug("entered - @_");
-
     # ignore volume
     $key = parse_neb_key($key);
+
+    my $log = $self->log;
+    my $sql = $self->sql;
+    my $db  = $self->db($key);
+
+    $log->debug("entered - @_");
 
     my $stat;
@@ -1272,7 +1540,10 @@
 }
 
-
+# this should have a 'db_index' as an argument
 sub mounts
 {
+    # XXX: this will only pull the mounts from one db
+    # XXX: loop over db_index and generate a single unique list
+    # XXX: or report mount info for each db server
     my $self = shift;
 
@@ -1281,5 +1552,5 @@
     my $log = $self->log;
     my $sql = $self->sql;
-    my $db  =$self->db;
+    my $db  = $self->_db_for_index(0); # XXX fix as above
 
     $log->debug("entered - @_");
@@ -1310,13 +1581,13 @@
     my $self = shift;
 
-    my $log = $self->log;
-    my $sql = $self->sql;
-    my $db  = $self->db;
+    my ($key, $name, $soft_volume) = @_;
+
+    my $log = $self->log;
+    my $sql = $self->sql;
+    my $db  = $self->db($key);
 
     no warnings qw( uninitialized );
     $log->debug( "entered - @_" );
     use warnings;
-
-    my ($name, $soft_volume) = @_;
 
     my ($vol_id, $vol_host, $vol_path, $xattr);
@@ -1335,5 +1606,5 @@
                 # find it, fall back to any volume
                 if ($soft_volume) {
-                    ($vol_id, $vol_host, $vol_path, $xattr) = $self->_get_storage_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
                 }
@@ -1375,15 +1646,13 @@
     my $self = shift;
 
-    my $log = $self->log;
-    my $sql = $self->sql;
-    my $db  =$self->db;
+    my $key = shift;
+
+    my $log = $self->log;
+    my $sql = $self->sql;
+    my $db  = $self->db($key);
 
     no warnings qw( uninitialized );
     $log->debug( "entered - @_" );
     use warnings;
-
-    my $key = shift;
-
-    $key = parse_neb_key($key);
 
     my ($vol_id, $vol_host, $vol_path, $xattr);
@@ -1423,9 +1692,9 @@
     my ($self, $key) = @_;
 
-    my $log = $self->log;
-    my $sql = $self->sql;
-    my $db  =$self->db;
-
     $key = parse_neb_key($key);
+
+    my $log = $self->log;
+    my $sql = $self->sql;
+    my $db  = $self->db($key);
 
     my $ext_id;
@@ -1452,11 +1721,12 @@
 sub _is_valid_volume_name
 {
-    my ($self, $vol_name) = @_;
-
-    my $log = $self->log;
-    my $sql = $self->sql;
-    my $db  =$self->db;
-
+    my ($self, $key, $vol_name) = @_;
+
+    $key = parse_neb_key($key);
     my $volume_info = parse_neb_volume($vol_name);
+
+    my $log = $self->log;
+    my $sql = $self->sql;
+    my $db  = $self->db($key);
 
     $vol_name = $volume_info->{volume};
@@ -1493,10 +1763,10 @@
     my $log = $self->log;
     my $sql = $self->sql;
-    my $db  = $self->db;
+    my $db  = $self->db($key);
 
     my $uri;
     eval {
-        my $storage_path = $self->_generate_storage_path($key, $vol_path);
-        my $storage_filename = $self->_generate_storage_filename($key, $ins_id);
+        my $storage_path = $self->_generate_storage_path($key->path, $vol_path);
+        my $storage_filename = $self->_generate_storage_filename($key->path, $ins_id);
         unless (-d $storage_path) {
             _retry(sub { mkpath(@_) }, $storage_path, 0, 0775)
@@ -1507,5 +1777,5 @@
         my $mode = [_retry(sub { stat($storage_path) } )]->[2] & 07777;
         unless ($mode == 0775) {
-            $log->error("$storage_path has the wrong permissions of: %04o", $mode);
+            $log->error("$storage_path has the wrong permissions of: %04x", $mode);
             _retry(sub { chmod(0775, $storage_path) }) or die "can't chmod $storage_path: $!";
         }
@@ -1523,5 +1793,5 @@
         my $path = $uri->file;
         die "can not set xattr on $path: $!"
-            unless (setfattr($path, 'user.nebulous_key', $key));
+            unless (setfattr($path, 'user.nebulous_key', $key->path));
     }
 
@@ -1579,9 +1849,4 @@
 
     my ($key, $vol_path) = @_;
-
-    # if the key has '/' in it, hash only the dirname() component
-    if ($key =~ m|/|) {
-        $key = dirname($key);
-    }
 
     # taken and modified from Cache::File::cache_file_path()
@@ -1627,11 +1892,12 @@
     my $log = $self->log;
     my $sql = $self->sql;
-    my $db  =$self->db;
+#    my $db  = $self->db;
 
     $log->debug( "entered" );
 
-    $self->db->disconnect;        
-
-    $log->debug( "disconnected from database: ", sub { $db->data_sources; } );
+# XXX do we need to loop over db_index?
+#    $self->db->disconnect;        
+
+#    $log->debug( "disconnected from database: ", sub { $db->data_sources; } );
 
     $log->debug( "leaving" );