Index: trunk/Nebulous-Server/Build.PL
===================================================================
--- trunk/Nebulous-Server/Build.PL	(revision 23699)
+++ trunk/Nebulous-Server/Build.PL	(revision 23932)
@@ -23,6 +23,6 @@
         'File::Temp'            => 0,
         'Filesys::Df'           => '0.92',
+        'Log::Dispatch::Email::MailSend' => 0,
         'Log::Log4perl'         => '0.48',
-        'Log::Dispatch::Email::MailSend' => 0,
         'Net::Server::Daemonize'=> '0.05',
         'Params::Validate'      => '0.73',
@@ -34,4 +34,5 @@
         'Test::More'            => '0.49',
         'Test::URI'             => '1.06',
+        'Test::DBUnit'          => '0.20',
     },
     recommends          => {
Index: trunk/Nebulous-Server/Changes
===================================================================
--- trunk/Nebulous-Server/Changes	(revision 23699)
+++ trunk/Nebulous-Server/Changes	(revision 23932)
@@ -8,4 +8,12 @@
     - add email logging of events to nebdiskd
     - attempt to optimize _is_valid_object_key() by eliminating an unused join
+    - rename Nebulous::Keys class -> Nebulous::Key
+    - select neb db to use by hashing only the directory component of the key
+    - disallow Nebulous::Server->rename_object() when it would cause the db
+      hash of a key to change
+    - create a pseduo directory structure on key creation
+    - Nebulous::Key parsing and testing improvements
+    - change 'log_level' param to 'trace'
+    - refactor ->find_objects() functionality
       
 0.16
Index: trunk/Nebulous-Server/MANIFEST
===================================================================
--- trunk/Nebulous-Server/MANIFEST	(revision 23699)
+++ trunk/Nebulous-Server/MANIFEST	(revision 23932)
@@ -4,5 +4,4 @@
 MANIFEST
 README
-Todo
 bin/neb-admin
 bin/neb-fsck
@@ -19,5 +18,5 @@
 examples/uri_test.pl
 init.d/nebdiskd
-lib/Nebulous/Keys.pm
+lib/Nebulous/Key.pm
 lib/Nebulous/Server.pm
 lib/Nebulous/Server.pod
@@ -33,4 +32,5 @@
 t/00_distribution.t 
 t/01_load.t
+t/02_config.t
 t/02_server_setup.t
 t/03_server_create_object.t
Index: trunk/Nebulous-Server/bin/neb-admin
===================================================================
--- trunk/Nebulous-Server/bin/neb-admin	(revision 23699)
+++ trunk/Nebulous-Server/bin/neb-admin	(revision 23932)
@@ -174,4 +174,8 @@
         # if the copies xattr is unset and the object has 2 or more instances, we
         # will assume a target of 2 copies
+	
+	# XXX change this: there should be no default value or we will
+	# have a race condition.  user.copies should never get set by
+	# any client until AFTER a file is no longer in use.
         my $copies = $obj->{copies} || 2;
 
Index: trunk/Nebulous-Server/lib/Nebulous/Key.pm
===================================================================
--- trunk/Nebulous-Server/lib/Nebulous/Key.pm	(revision 23932)
+++ trunk/Nebulous-Server/lib/Nebulous/Key.pm	(revision 23932)
@@ -0,0 +1,279 @@
+# Copyright (c) 2004  Joshua Hoblitt
+#
+# $Id: Keys.pm,v 1.3 2008-09-09 02:18:47 jhoblitt Exp $
+
+package Nebulous::Key;
+
+use strict;
+use warnings FATAL => qw( all );
+
+our $VERSION = '0.02';
+
+use base qw( Exporter Class::Accessor::Fast );
+
+use File::Spec;
+use URI::file;
+use URI;
+use overload '""' => \&_stringify_key;
+
+our @EXPORT_OK = qw(
+    parse_neb_key
+    parse_neb_volume
+);
+
+__PACKAGE__->mk_ro_accessors(qw( path volume soft_volume ));
+
+sub parse_neb_key
+{
+    my ($key, $volume) = @_;
+    return unless defined $key;
+
+    # white space is not allowed
+    if ($key =~ qr/\s+/) {
+        die "keys and URIs may not contain whitespace";
+    }
+
+    my $uri = URI->new($key);
+    my $scheme = $uri->scheme;
+    my $path = $uri->opaque;
+    
+    my $volume_name;
+    my $soft_volume;
+    # if this is a valid uri 
+    if (defined $scheme) {
+        # if so, does it use the neb scheme?
+        die "URI does not use the 'neb' scheme"
+            unless $scheme eq 'neb'; 
+
+        # neb:path (not allowed)
+        # neb://<volume name>/...
+        # neb:/path... (leading '/' is stripped)
+        # neb:///path... (same as neb:/path)
+
+        # volume specifier must be at least one character
+        # strip off volume component if it exists
+        $path =~ s|^//([^/]+)||;
+
+        # a new is not allowed to be just a volume specifier, it must have a
+        # path component to it
+        unless (length $path) {
+            die "neb URI scheme requires a path component"; 
+        }
+        
+        # ignore key supplied volume name if one is passed as a parameter
+        $volume ||= $1;
+        my $volume_info = parse_neb_volume($volume);
+
+        # magically eat the "any" volume
+        if (defined $volume_info->{volume} and $volume_info->{volume} ne 'any') {
+            $volume_name = $volume_info->{volume};
+            $soft_volume = $volume_info->{soft_volume};
+        }
+
+        # require a leading slash if there is no volume name
+        if ((not defined $volume_name) and (not $path =~ m|^/|)) {
+            die "neb URI scheme requires a leading slash, eg. neb:/";
+        }
+    } else {
+        my $volume_info = parse_neb_volume($volume);
+
+        # magically eat the "any" volume
+        if (defined $volume_info->{volume} and $volume_info->{volume} ne 'any') {
+            $volume_name = $volume_info->{volume};
+            $soft_volume = $volume_info->{soft_volume};
+        }
+    }
+
+    # strip leading slashes
+    $path =~ s|^/+||;
+
+    # strip leading '.' or '..'
+    $path =~ s|^\.{1,2}||;
+
+    # remove multiple /'s and trailing slashes
+    $path = File::Spec->canonpath($path);
+
+    return __PACKAGE__->new({
+        volume      => $volume_name,
+        soft_volume => $soft_volume,
+        path        => $path,
+    });
+}
+
+
+sub parse_neb_volume
+{
+    my $volume = shift;
+    return unless defined $volume;
+
+    my $soft_volume;
+    # check to see if there is a tilde and remove it if found
+    unless (defined $volume and $volume =~ s/^~//) {
+        $soft_volume = 1;
+    }
+
+    return({ volume => $volume, soft_volume => $soft_volume });
+}
+
+
+sub _stringify_key
+{
+    my $self = shift;
+
+    my $path        = $self->path;
+    my $volume      = $self->volume || "";
+    my $soft_volume = $self->soft_volume ? '~' : "";
+
+    return "neb://${soft_volume}${volume}/$path";
+}
+
+
+1;
+
+__END__
+
+=pod
+
+=head1 NAME
+
+Nebulous::Key - Nebulous Keys Explained
+
+=head1 DESCRIPTION
+
+The Nebulous system interprets it's storage object keys in a I<slightly>
+magical manner.  It supports both plain vanilla C<key strings> and keys in the
+form of an L<URI>.  In the L<URI> form a specific storage volume can be implied
+(as a request, not a requirement).  Both key forms are subject to mangling such
+that I<IF> the key I<looks> like a C<POSIX> semantics file path, it is
+canocalized.
+
+=head1 RESERVED SYNTAX
+
+This particular syntax is disallowed and is reserved for future use.
+
+    <neb:<phrase>/<path>
+
+=head1 EXAMPLES
+
+=head2 Key Strings
+
+    key:        foo/bar/baz/quix
+    mangled to: foo/bar/baz/quix
+    volume:     any
+
+    key:        /foo/bar/baz/quix
+    mangled to: foo/bar/baz/quix
+    volume:     any
+
+    key:        //foo/bar/baz/quix
+    mangled to: foo/bar/baz/quix
+    volume:     any
+
+    key:        ///foo/bar/baz/quix
+    mangled to: foo/bar/baz/quix
+    volume:     any
+
+    key:        foo////bar/baz/quix
+    mangled to: foo/bar/baz/quix
+    volume:     any
+
+    key:        foo/bar/baz/quix/
+    mangled to: foo/bar/baz/quix
+    volume:     any
+
+=head2 URI w/ volume name
+
+neb://<volume>/<path>
+
+    key:        neb://foo/bar/baz/quix
+    mangled to: bar/baz/quix
+    volume:     foo
+
+    key:        neb://foo//bar/baz/quix
+    mangled to: bar/baz/quix
+    volume:     foo
+
+    key:        neb://foo///bar/baz/quix
+    mangled to: bar/baz/quix
+    volume:     foo
+
+    key:        neb://foo/bar///baz/quix
+    mangled to: bar/baz/quix
+    volume:     foo
+
+    key:        neb://foo/bar/baz/quix/
+    mangled to: bar/baz/quix
+    volume:     foo
+
+=head2 URI w/o volume name
+
+neb:/<path>
+
+    key:        neb:/foo/bar/baz/quix
+    mangled to: foo/bar/baz/quix
+    volume:     any
+
+    key:        neb:///foo/bar/baz/quix
+    mangled to: foo/bar/baz/quix
+    volume:     any
+
+    key:        neb://///foo/bar/baz/quix
+    mangled to: foo/bar/baz/quix
+    volume:     any
+
+=head2 Forbidden Keys
+
+=head3 Key with whitespace
+
+    "/ foo/bar/baz/quix"
+    " /foo/bar/baz/quix"
+    "/foo/bar/baz/quix "
+
+=head3 URI with whitespace
+
+    "neb ://foo/bar/baz/quix"
+    "neb:// foo/bar/baz/quix"
+    " neb://foo/bar/baz/quix"
+    "neb://foo/bar/baz/quix "
+
+=head3 URI with out a volume requires a leading slash
+
+    neb:foo/bar/baz/quix
+
+=head1 CREDITS
+
+Just me, myself, and I.
+
+=head1 SUPPORT
+
+Please contact the author directly via e-mail.
+
+=head1 AUTHOR
+
+Joshua Hoblitt <jhoblitt@cpan.org>
+
+=head1 COPYRIGHT
+
+Copyright (C) 2008  Joshua Hoblitt.  All rights reserved.
+
+This program is free software; you can redistribute it and/or modify it under
+the terms of the GNU General Public License as published by the Free Software
+Foundation; either version 2 of the License, or (at your option) any later
+version.
+
+This program is distributed in the hope that it will be useful, but WITHOUT ANY
+WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
+PARTICULAR PURPOSE.  See the GNU General Public License for more details.
+
+You should have received a copy of the GNU General Public License along with
+this program; if not, write to the Free Software Foundation, Inc., 59 Temple
+Place - Suite 330, Boston, MA  02111-1307, USA.
+
+The full text of the license can be found in the LICENSE file included with
+this module, or in the L<perlgpl> Pod as supplied with Perl 5.8.1 and later.
+
+=head1 SEE ALSO
+
+L<Nebulous::Server>, L<Nebulous::Client>, L<nebclient(3)>
+
+=cut
Index: trunk/Nebulous-Server/lib/Nebulous/Keys.pm
===================================================================
--- trunk/Nebulous-Server/lib/Nebulous/Keys.pm	(revision 23699)
+++ 	(revision )
@@ -1,267 +1,0 @@
-# Copyright (c) 2004  Joshua Hoblitt
-#
-# $Id: Keys.pm,v 1.3 2008-09-09 02:18:47 jhoblitt Exp $
-
-package Nebulous::Keys;
-
-use strict;
-use warnings FATAL => qw( all );
-
-our $VERSION = '0.02';
-
-use base qw( Exporter Class::Accessor::Fast );
-
-use File::Spec;
-use URI::file;
-use URI;
-use overload '""' => \&_stringify_key;
-
-our @EXPORT_OK = qw(
-    parse_neb_key
-    parse_neb_volume
-);
-
-__PACKAGE__->mk_ro_accessors(qw( path volume soft_volume ));
-
-sub parse_neb_key
-{
-    my ($key, $volume) = @_;
-    return unless defined $key;
-
-    # white space is not allowed
-    if ($key =~ qr/\s+/) {
-        die "keys and URIs may not contain whitespace";
-    }
-
-    my $uri = URI->new($key);
-    my $scheme = $uri->scheme;
-    my $path = $uri->opaque;
-    
-    my $volume_name;
-    my $soft_volume;
-    # if this is a valid uri 
-    if (defined $scheme) {
-        # if so, does it use the neb scheme?
-        die "URI does not use the 'neb' scheme"
-            unless $scheme eq 'neb'; 
-
-        # neb:path (not allowed)
-        # neb://<volume name>/...
-        # neb:/path... (leading '/' is stripped)
-        # neb:///path... (same as neb:/path)
-
-        # volume specifier must be at least one character
-        # strip off volume component if it exists
-        $path =~ s|^//([^/]+)||;
-        
-        # ignore key supplied volume name if one is passed as a parameter
-        $volume ||= $1;
-        my $volume_info = parse_neb_volume($volume);
-
-        # magically eat the "any" volume
-        if (defined $volume_info->{volume} and $volume_info->{volume} ne 'any') {
-            $volume_name = $volume_info->{volume};
-            $soft_volume = $volume_info->{soft_volume};
-        }
-
-        # require a leading slash if there is no volume name
-        if ((not defined $volume_name) and (not $path =~ m|^/|)) {
-            die "neb URI scheme requires a leading slash, eg. neb:/";
-        }
-    } else {
-        my $volume_info = parse_neb_volume($volume);
-
-        # magically eat the "any" volume
-        if (defined $volume_info->{volume} and $volume_info->{volume} ne 'any') {
-            $volume_name = $volume_info->{volume};
-            $soft_volume = $volume_info->{soft_volume};
-        }
-    }
-
-    # strip leading slashes
-    $path =~ s|^/+||;
-
-    # remove multiple /'s and trailing slashes
-    $path = File::Spec->canonpath($path);
-
-    return __PACKAGE__->new({
-        volume      => $volume_name,
-        soft_volume => $soft_volume,
-        path        => $path,
-    });
-}
-
-sub parse_neb_volume
-{
-    my $volume = shift;
-    return unless defined $volume;
-
-    my $soft_volume;
-    # check to see if there is a tilde and remove it if found
-    unless (defined $volume and $volume =~ s/^~//) {
-        $soft_volume = 1;
-    }
-
-    return({ volume => $volume, soft_volume => $soft_volume });
-}
-
-sub _stringify_key
-{
-    my $self = shift;
-
-    my $path        = $self->path;
-    my $volume      = $self->volume || "";
-    my $soft_volume = $self->soft_volume ? '~' : "";
-
-    return "neb://${soft_volume}${volume}/$path";
-}
-
-1;
-
-__END__
-
-=pod
-
-=head1 NAME
-
-Nebulous::Keys - Nebulous Keys Explained
-
-=head1 DESCRIPTION
-
-The Nebulous system interprets it's storage object keys in a I<slightly>
-magical manner.  It supports both plain vanilla C<key strings> and keys in the
-form of an L<URI>.  In the L<URI> form a specific storage volume can be implied
-(as a request, not a requirement).  Both key forms are subject to mangling such
-that I<IF> the key I<looks> like a C<POSIX> semantics file path, it is
-canocalized.
-
-=head1 RESERVED SYNTAX
-
-This particular syntax is disallowed and is reserved for future use.
-
-    <neb:<phrase>/<path>
-
-=head1 EXAMPLES
-
-=head2 Key Strings
-
-    key:        foo/bar/baz/quix
-    mangled to: foo/bar/baz/quix
-    volume:     any
-
-    key:        /foo/bar/baz/quix
-    mangled to: foo/bar/baz/quix
-    volume:     any
-
-    key:        //foo/bar/baz/quix
-    mangled to: foo/bar/baz/quix
-    volume:     any
-
-    key:        ///foo/bar/baz/quix
-    mangled to: foo/bar/baz/quix
-    volume:     any
-
-    key:        foo////bar/baz/quix
-    mangled to: foo/bar/baz/quix
-    volume:     any
-
-    key:        foo/bar/baz/quix/
-    mangled to: foo/bar/baz/quix
-    volume:     any
-
-=head2 URI w/ volume name
-
-neb://<volume>/<path>
-
-    key:        neb://foo/bar/baz/quix
-    mangled to: bar/baz/quix
-    volume:     foo
-
-    key:        neb://foo//bar/baz/quix
-    mangled to: bar/baz/quix
-    volume:     foo
-
-    key:        neb://foo///bar/baz/quix
-    mangled to: bar/baz/quix
-    volume:     foo
-
-    key:        neb://foo/bar///baz/quix
-    mangled to: bar/baz/quix
-    volume:     foo
-
-    key:        neb://foo/bar/baz/quix/
-    mangled to: bar/baz/quix
-    volume:     foo
-
-=head2 URI w/o volume name
-
-neb:/<path>
-
-    key:        neb:/foo/bar/baz/quix
-    mangled to: foo/bar/baz/quix
-    volume:     any
-
-    key:        neb:///foo/bar/baz/quix
-    mangled to: foo/bar/baz/quix
-    volume:     any
-
-    key:        neb://///foo/bar/baz/quix
-    mangled to: foo/bar/baz/quix
-    volume:     any
-
-=head2 Forbidden Keys
-
-=head3 Key with whitespace
-
-    "/ foo/bar/baz/quix"
-    " /foo/bar/baz/quix"
-    "/foo/bar/baz/quix "
-
-=head3 URI with whitespace
-
-    "neb ://foo/bar/baz/quix"
-    "neb:// foo/bar/baz/quix"
-    " neb://foo/bar/baz/quix"
-    "neb://foo/bar/baz/quix "
-
-=head3 URI with out a volume requires a leading slash
-
-    neb:foo/bar/baz/quix
-
-=head1 CREDITS
-
-Just me, myself, and I.
-
-=head1 SUPPORT
-
-Please contact the author directly via e-mail.
-
-=head1 AUTHOR
-
-Joshua Hoblitt <jhoblitt@cpan.org>
-
-=head1 COPYRIGHT
-
-Copyright (C) 2008  Joshua Hoblitt.  All rights reserved.
-
-This program is free software; you can redistribute it and/or modify it under
-the terms of the GNU General Public License as published by the Free Software
-Foundation; either version 2 of the License, or (at your option) any later
-version.
-
-This program is distributed in the hope that it will be useful, but WITHOUT ANY
-WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
-PARTICULAR PURPOSE.  See the GNU General Public License for more details.
-
-You should have received a copy of the GNU General Public License along with
-this program; if not, write to the Free Software Foundation, Inc., 59 Temple
-Place - Suite 330, Boston, MA  02111-1307, USA.
-
-The full text of the license can be found in the LICENSE file included with
-this module, or in the L<perlgpl> Pod as supplied with Perl 5.8.1 and later.
-
-=head1 SEE ALSO
-
-L<Nebulous::Server>, L<Nebulous::Client>, L<nebclient(3)>
-
-=cut
Index: trunk/Nebulous-Server/lib/Nebulous/Server.pm
===================================================================
--- trunk/Nebulous-Server/lib/Nebulous/Server.pm	(revision 23699)
+++ 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" );
Index: trunk/Nebulous-Server/lib/Nebulous/Server.pod
===================================================================
--- trunk/Nebulous-Server/lib/Nebulous/Server.pod	(revision 23699)
+++ trunk/Nebulous-Server/lib/Nebulous/Server.pod	(revision 23932)
@@ -26,5 +26,5 @@
     Nebulous::Server->find_objects( $pattern );
     Nebulous::Server->find_instances( $key, $volume );
-    Nebulous::Server->delete_instance( $uri );
+    Nebulous::Server->delete_instance( $key, $uri );
     Nebulous::Server->stat_object( $key );
     Nebulous::Server->mounts();
@@ -160,8 +160,8 @@
 Throws an exception on error.
 
-=item * delete_instance( $uri );
-
-Accepts 1 parameters, it is mandatory.  Returns Boolean true.  Throws an
-exception on error.
+=item * delete_instance( $key, $uri );
+
+Accepts 2 parameters, both mandatory.  Returns Boolean true.  C<<$uri>> must be
+an instance of C<<$key>>.  Throws an exception on error.
 
 =item * swap_objects( $key1, $key2 );
Index: trunk/Nebulous-Server/lib/Nebulous/Server/Config.pm
===================================================================
--- trunk/Nebulous-Server/lib/Nebulous/Server/Config.pm	(revision 23699)
+++ trunk/Nebulous-Server/lib/Nebulous/Server/Config.pm	(revision 23932)
@@ -8,10 +8,10 @@
 use warnings FATAL => qw( all );
 
-our $VERSION = 0.02;
+our $VERSION = 0.03;
 
 use base qw( Class::Accessor::Fast );
 
 use Log::Log4perl qw( :levels );
-use Params::Validate qw( validate SCALAR );
+use Params::Validate qw( validate validate_pos SCALAR );
 
 our %LEVELS = (
@@ -25,12 +25,19 @@
 );
 
-my $new_validate = {
+my $db_validate = {
+    dbindex       => {
+        type => SCALAR,
+        regex => qr/^\d+$/,
+    },
     dsn         => { type => SCALAR },
     dbuser      => { type => SCALAR },
     dbpasswd    => { type => SCALAR },
-    log_level   => {
+};
+
+my $new_validate = {
+    trace       => {
         type        => SCALAR,
         optional    => 1,
-        default     => 'all',
+        default     => 'fatal',
         callbacks   => {
             'is valid level' => sub {
@@ -39,9 +46,14 @@
         },
     },
+    dsn         => { type => SCALAR, optional => 1 },
+    dbuser      => { type => SCALAR, optional => 1 },
+    dbpasswd    => { type => SCALAR, optional => 1 },
 };
 
 __PACKAGE__->mk_ro_accessors( keys %$new_validate );
 
-sub init {
+
+sub new
+{
     my $class = shift;
 
@@ -49,13 +61,74 @@
 
     # normalize log levels to lower-case
-    $p{ log_level } = lc $p{ log_level };
-
-    my $self = \%p;
+    my $self = { trace => $LEVELS{lc($p{trace})} };
 
     bless $self, $class || ref $class;
+
+    my @dbs;
+    $self->{dbs} = \@dbs;
+
+    if (defined $p{dsn} or defined $p{dbuser} or defined $p{dbpasswd}) {
+        $self->add_db(
+            dbindex     => 0,
+            dsn         => $p{dsn},
+            dbuser      => $p{dbuser},
+            dbpasswd    => $p{dbpasswd},
+        );
+    }
 
     return $self;
 }
 
+
+sub add_db
+{
+    my $self = shift;
+    
+    my %p = validate( @_, $db_validate );
+
+    my $config_db = Nebulous::Server::Config::DB->new({
+        dsn         => $p{dsn},
+        dbuser      => $p{dbuser},
+        dbpasswd    => $p{dbpasswd},
+    });
+
+    $self->{dbs}->[$p{dbindex}] = $config_db;
+
+    return $self;
+}
+
+
+sub db 
+{
+    my $self = shift;
+
+    my ($db_index) = validate_pos( @_, { type => SCALAR, optional => 1, });
+
+    # default to 0
+    $db_index ||= 0;
+
+    return $self->{dbs}->[$db_index];
+}
+
+
+sub n_db 
+{
+    my $self = shift;
+
+    return scalar @{ $self->{dbs} };
+}
+
+
+package Nebulous::Server::Config::DB;
+
+use strict;
+use warnings FATAL => qw( all );
+
+our $VERSION = 0.01;
+
+use base qw( Class::Accessor::Fast );
+
+__PACKAGE__->mk_ro_accessors(qw( dsn dbuser dbpasswd )); 
+
 1;
 
Index: trunk/Nebulous-Server/lib/Nebulous/Server/Log.pm
===================================================================
--- trunk/Nebulous-Server/lib/Nebulous/Server/Log.pm	(revision 23699)
+++ trunk/Nebulous-Server/lib/Nebulous/Server/Log.pm	(revision 23932)
@@ -16,7 +16,7 @@
     my ($self, $config) = @_;
 
-    my $dsn         = $config->dsn;
-    my $dbuser      = $config->dbuser;
-    my $dbpasswd    = $config->dbpasswd;
+#    my $dsn         = $config->db->dsn;
+#    my $dbuser      = $config->db->dbuser;
+#    my $dbpasswd    = $config->db->dbpasswd;
 
     my $conf = <<END;
@@ -30,9 +30,9 @@
 #   date | hostname | priority | method/sub - message\n
 
-    log4perl.appender.SQLLOG            = Log::Log4perl::Appender::DBI
-    log4perl.appender.SQLLOG.datasource = $dsn
-    log4perl.appender.SQLLOG.username   = $dbuser
-    log4perl.appender.SQLLOG.password   = $dbpasswd
-    log4perl.appender.SQLLOG.sql        = \
+#    log4perl.appender.SQLLOG            = Log::Log4perl::Appender::DBI
+#    log4perl.appender.SQLLOG.sql        = \
+#    log4perl.appender.SQLLOG.datasource = 
+#    log4perl.appender.SQLLOG.username   = 
+#    log4perl.appender.SQLLOG.password   = 
     INSERT INTO log (timestamp, hostname, level, sub, message) \
     VALUES          (%d,        %H,       %p,    %M,  %m)
Index: trunk/Nebulous-Server/lib/Nebulous/Server/SOAP.pm
===================================================================
--- trunk/Nebulous-Server/lib/Nebulous/Server/SOAP.pm	(revision 23699)
+++ trunk/Nebulous-Server/lib/Nebulous/Server/SOAP.pm	(revision 23932)
@@ -1,5 +1,5 @@
 # Copyright (c) 2004  Joshua Hoblitt
 #
-# $Id: SOAP.pm,v 1.6 2008-12-14 22:54:25 eugene Exp $
+# $Id: SOAP.pm,v 1.4.32.1 2008-12-14 22:52:37 eugene Exp $
 
 package Nebulous::Server::SOAP;
@@ -17,8 +17,10 @@
 our $AUTOLOAD;
 
-our @args;
+our $config;
 our $neb;
 
-sub new_on_init {
+
+sub new_on_init
+{
     my $self = shift;
 
@@ -27,5 +29,5 @@
     require Apache2::ServerUtil;
 
-    @args = @_;
+    $config = shift;
 
     my $s = Apache2::ServerUtil->server;
@@ -35,13 +37,17 @@
 }
 
-sub init {
+
+sub init
+{
     my $self = shift;
 
-    $neb = Nebulous::Server->new(@args);        
+    $neb = Nebulous::Server->new_from_config($config);        
 
     return Apache2::Const::OK;
 }
 
-sub AUTOLOAD {
+
+sub AUTOLOAD
+{
     my $self = shift;
 
@@ -76,3 +82,4 @@
 }
 
+
 1;
Index: trunk/Nebulous-Server/lib/Nebulous/Server/SQL.pm
===================================================================
--- trunk/Nebulous-Server/lib/Nebulous/Server/SQL.pm	(revision 23699)
+++ trunk/Nebulous-Server/lib/Nebulous/Server/SQL.pm	(revision 23932)
@@ -8,5 +8,5 @@
 use warnings FATAL => qw( all );
 
-our $VERSION = '0.02';
+our $VERSION = '0.04';
 
 use base qw( Class::Accessor::Fast );
@@ -26,6 +26,6 @@
     new_object          => qq{
         INSERT INTO storage_object
-        (so_id, ext_id, type)
-        VALUES (?, ?, 'REG_FILE')
+        (so_id, ext_id, ext_id_basename, type, dir_id)
+        VALUES (?, ?, ?, 'REG_FILE', ?)
     },
     new_object_attr  => qq{
@@ -60,4 +60,16 @@
         USING (so_id)
         WHERE ext_id = ?
+    },
+    get_directory       => qq{
+        SELECT
+            dir_id
+        FROM directory
+        WHERE parent_id = ?
+            AND dirname = ?
+    },
+    new_directory       => qq{
+        INSERT INTO directory
+        (dirname, parent_id)
+        VALUES (?, ?)
     },
     check_object_name => qq{
@@ -297,12 +309,17 @@
             USING(vol_id)
     },
-    find_objects => qq{
-        SELECT *
-        FROM storage_object
-        WHERE ext_id REGEXP ?
+    find_object_by_ext_id => qq{
+        SELECT ext_id, ext_id_basename
+        FROM storage_object
+        WHERE ext_id = ?
+    },
+    find_object_by_dir_id => qq{
+        SELECT ext_id, ext_id_basename
+        FROM storage_object
+        WHERE dir_id = ?
     },
     rename_object => qq{
         UPDATE storage_object
-        SET ext_id = ?
+        SET ext_id = ?, ext_id_basename = ?, dir_id = ?
         WHERE ext_id = ?
     },
@@ -330,4 +347,27 @@
         GROUP BY so_id
         HAVING available_instances < instances OR instances < copies
+    },
+    find_objects_with_extra_instances_by_xattr => qq{
+        SELECT
+            so.so_id,
+            so.ext_id,
+            count(ins_id) as instances,
+            mv.name as volume_name,
+            mv.host as volume_host,
+            count(mv.vol_id) as available_instances,
+            xattr.value as copies
+        FROM storage_object AS so
+        JOIN storage_object_xattr as xattr
+            ON so.so_id = xattr.so_id
+            AND xattr.name = 'user.copies'
+        JOIN instance AS i
+            ON so.so_id = i.so_id
+        JOIN mountedvol AS mv
+            USING(vol_id)
+        WHERE
+            mv.available = 1
+        GROUP BY so_id
+        HAVING available_instances > copies
+        limit 5;
     },
     find_objects_with_extra_instances => qq{
@@ -387,4 +427,5 @@
 DROP TABLE IF EXISTS log;
 DROP TABLE IF EXISTS mountedvol;
+DROP TABLE IF EXISTS directory;
 DROP PROCEDURE IF EXISTS getmountedvol;
 SET FOREIGN_KEY_CHECKS=1
@@ -408,7 +449,25 @@
 
 __DATA__
+CREATE TABLE directory (
+    dir_id BIGINT NOT NULL AUTO_INCREMENT,
+    dirname CHAR(255) NOT NULL,
+    parent_id BIGINT NOT NULL,
+    FOREIGN KEY(parent_id) REFERENCES directory(dir_id),
+    PRIMARY KEY(dir_id),
+    KEY(parent_id)
+) ENGINE=innodb DEFAULT CHARSET=latin1;
+
+###
+
+INSERT INTO directory (dir_id, dirname, parent_id) VALUES (1, '/', 1);
+
+###
+
 CREATE TABLE storage_object (
     so_id BIGINT NOT NULL AUTO_INCREMENT,
     ext_id VARCHAR(255) NOT NULL UNIQUE,
+    ext_id_basename VARCHAR(255) NOT NULL,
+    dir_id BIGINT NOT NULL,
+    FOREIGN KEY(dir_id) REFERENCES directory(dir_id),
     type enum('REG_FILE'),
     PRIMARY KEY(so_id),
Index: trunk/Nebulous-Server/t/01_load.t
===================================================================
--- trunk/Nebulous-Server/t/01_load.t	(revision 23699)
+++ trunk/Nebulous-Server/t/01_load.t	(revision 23932)
@@ -12,5 +12,5 @@
 use Test::More tests => 5;
 
-BEGIN { use_ok( 'Nebulous::Keys' ); }
+BEGIN { use_ok( 'Nebulous::Key' ); }
 BEGIN { use_ok( 'Nebulous::Server' ); }
 BEGIN { use_ok( 'Nebulous::Server::Log' ); }
Index: trunk/Nebulous-Server/t/02_config.t
===================================================================
--- trunk/Nebulous-Server/t/02_config.t	(revision 23699)
+++ trunk/Nebulous-Server/t/02_config.t	(revision 23932)
@@ -3,5 +3,5 @@
 # Copryight (C) 2004-2005  Joshua Hoblitt
 #
-# $Id: 02_config.t,v 1.1 2008-12-12 21:13:41 jhoblitt Exp $
+# $Id: 02_config.t,v 1.1.2.2 2008-12-14 22:52:37 eugene Exp $
 
 use strict;
Index: trunk/Nebulous-Server/t/02_server_setup.t
===================================================================
--- trunk/Nebulous-Server/t/02_server_setup.t	(revision 23699)
+++ trunk/Nebulous-Server/t/02_server_setup.t	(revision 23932)
@@ -8,5 +8,5 @@
 use warnings;
 
-use Test::More tests => 2;
+use Test::More tests => 6;
 
 use lib qw( ./t ./lib );
@@ -17,23 +17,49 @@
 Test::Nebulous->setup;
 
-isa_ok(
-    Nebulous::Server->new(
-        dsn         => $NEB_DB,
-        dbuser      => $NEB_USER,
-        dbpasswd    => $NEB_PASS,
-    ),
-    "Nebulous::Server"
-);
+isa_ok(Nebulous::Server->new(), "Nebulous::Server");
 
 Test::Nebulous->setup;
 
-eval {
-    Nebulous::Server->new(
+# ->new()
+{
+    my $neb = Nebulous::Server->new( trace => 'off' );
+
+    ok($neb, "set log level");
+}
+
+Test::Nebulous->setup;
+
+{
+    my $neb = Nebulous::Server->new(
         dsn         => "DBI:mysql:database=foobar:host=localhost",
         dbuser      => "baz",
         dbpasswd    =>"boo",
     );
-};
-like( $@, qr/DBI connect.*? failed/, "bad dsn/user/pass" );
+
+    ok($neb, "set database");
+}
+
+Test::Nebulous->setup;
+
+# add dbs after ->new()
+{
+    my $neb = Nebulous::Server->new;
+
+    ok($neb->config->add_db(
+        dbindex    => 0,
+        dsn         => "DBI:mysql:database=foobar:host=localhost",
+        dbuser      => "baz",
+        dbpasswd    =>"boo",
+    ), "add db");
+    
+    ok($neb->config->add_db(
+        dbindex    => 1,
+        dsn         => "DBI:mysql:database=foobar:host=localhost",
+        dbuser      => "baz",
+        dbpasswd    =>"boo",
+    ), "add dbs");
+
+    is($neb->config->n_db, 2, "n dbs")
+}
 
 Test::Nebulous->cleanup;
Index: trunk/Nebulous-Server/t/03_server_create_object.t
===================================================================
--- trunk/Nebulous-Server/t/03_server_create_object.t	(revision 23699)
+++ trunk/Nebulous-Server/t/03_server_create_object.t	(revision 23932)
@@ -8,8 +8,9 @@
 use warnings FATAL => qw( all );
 
-use Test::More tests => 89;
+use Test::More tests => 99;
 
 use lib qw( ./t ./lib );
 
+use File::Basename qw( basename );
 use File::ExtAttr qw( getfattr );
 use Nebulous::Server;
@@ -26,4 +27,6 @@
 );
 
+use Test::DBUnit dsn => $NEB_DB, username => $NEB_USER, password => $NEB_PASS;
+
 Test::Nebulous->setup;
 
@@ -334,4 +337,174 @@
 }
 
+# test for properly row creation in the directories table
+
+Test::Nebulous->setup;
+
+{
+    my $key = "foo";
+    $neb->create_object($key);
+
+    expected_dataset_ok(
+        directory       => [dir_id => 1, dirname => '/', parent_id => 1],
+        storage_object  => [so_id => 1, ext_id => $key, dir_id => 1],
+    );
+}
+
+Test::Nebulous->setup;
+
+{
+    my $key = "a/foo";
+    $neb->create_object($key);
+
+    expected_dataset_ok(
+        directory       => [dir_id => 1, dirname => '/', parent_id => 1],
+        directory       => [dir_id => 2, dirname => 'a', parent_id => 1],
+        storage_object  => [so_id => 1, ext_id => $key, ext_id_basename => basename($key), dir_id => 2],
+    );
+
+}
+
+Test::Nebulous->setup;
+
+{
+    my $key = "a/b/foo";
+    $neb->create_object($key);
+
+    expected_dataset_ok(
+        directory       => [dir_id => 1, dirname => '/', parent_id => 1],
+        directory       => [dir_id => 2, dirname => 'a', parent_id => 1],
+        directory       => [dir_id => 3, dirname => 'b', parent_id => 2],
+        storage_object  => [so_id => 1, ext_id => $key, ext_id_basename => basename($key), dir_id => 3],
+    );
+}
+
+Test::Nebulous->setup;
+
+{
+    my $key = "a/b/c/foo";
+    $neb->create_object($key);
+
+    expected_dataset_ok(
+        directory       => [dir_id => 1, dirname => '/', parent_id => 1],
+        directory       => [dir_id => 2, dirname => 'a', parent_id => 1],
+        directory       => [dir_id => 3, dirname => 'b', parent_id => 2],
+        directory       => [dir_id => 4, dirname => 'c', parent_id => 3],
+        storage_object  => [so_id => 1, ext_id => $key, ext_id_basename => basename($key), dir_id => 4],
+    );
+}
+
+Test::Nebulous->setup;
+
+{
+    my $key1 = "a/b/c/foo";
+    my $key2 = "foo";
+    $neb->create_object($key1);
+    $neb->create_object($key2);
+
+    expected_dataset_ok(
+        directory       => [dir_id => 1, dirname => '/', parent_id => 1],
+        storage_object  => [so_id => 2, ext_id => $key2, ext_id_basename => basename($key2), dir_id => 1],
+        directory       => [dir_id => 2, dirname => 'a', parent_id => 1],
+        directory       => [dir_id => 3, dirname => 'b', parent_id => 2],
+        directory       => [dir_id => 4, dirname => 'c', parent_id => 3],
+        storage_object  => [so_id => 1, ext_id => $key1, ext_id_basename => basename($key1), dir_id => 4],
+    );
+}
+
+Test::Nebulous->setup;
+
+{
+    my $key1 = "a/b/c/foo";
+    my $key2 = "a/foo";
+    $neb->create_object($key1);
+    $neb->create_object($key2);
+
+    expected_dataset_ok(
+        directory       => [dir_id => 1, dirname => '/', parent_id => 1],
+        directory       => [dir_id => 2, dirname => 'a', parent_id => 1],
+        storage_object  => [so_id => 2, ext_id => $key2, ext_id_basename => basename($key2), dir_id => 2],
+        directory       => [dir_id => 3, dirname => 'b', parent_id => 2],
+        directory       => [dir_id => 4, dirname => 'c', parent_id => 3],
+        storage_object  => [so_id => 1, ext_id => $key1, ext_id_basename => basename($key1), dir_id => 4],
+    );
+}
+
+Test::Nebulous->setup;
+
+{
+    my $key1 = "a/b/c/foo";
+    my $key2 = "d/foo";
+    $neb->create_object($key1);
+    $neb->create_object($key2);
+
+    expected_dataset_ok(
+        directory       => [dir_id => 1, dirname => '/', parent_id => 1],
+        directory       => [dir_id => 2, dirname => 'a', parent_id => 1],
+        directory       => [dir_id => 3, dirname => 'b', parent_id => 2],
+        directory       => [dir_id => 4, dirname => 'c', parent_id => 3],
+        storage_object  => [so_id => 1, ext_id => $key1, ext_id_basename => basename($key1), dir_id => 4],
+        directory       => [dir_id => 5, dirname => 'd', parent_id => 1],
+        storage_object  => [so_id => 2, ext_id => $key2, ext_id_basename => basename($key2), dir_id => 5],
+    );
+}
+
+Test::Nebulous->setup;
+
+{
+    my $key1 = "a/b/c/foo";
+    my $key2 = "a/d/foo";
+    $neb->create_object($key1);
+    $neb->create_object($key2);
+
+    expected_dataset_ok(
+        directory       => [dir_id => 1, dirname => '/', parent_id => 1],
+        directory       => [dir_id => 2, dirname => 'a', parent_id => 1],
+        directory       => [dir_id => 3, dirname => 'b', parent_id => 2],
+        directory       => [dir_id => 4, dirname => 'c', parent_id => 3],
+        storage_object  => [so_id => 1, ext_id => $key1, ext_id_basename => basename($key1), dir_id => 4],
+        directory       => [dir_id => 5, dirname => 'd', parent_id => 2],
+        storage_object  => [so_id => 2, ext_id => $key2, ext_id_basename => basename($key2), dir_id => 5],
+    );
+}
+
+Test::Nebulous->setup;
+
+{
+    my $key1 = "a/b/c/foo";
+    my $key2 = "d/a/foo";
+    $neb->create_object($key1);
+    $neb->create_object($key2);
+
+    expected_dataset_ok(
+        directory       => [dir_id => 1, dirname => '/', parent_id => 1],
+        directory       => [dir_id => 2, dirname => 'a', parent_id => 1],
+        directory       => [dir_id => 3, dirname => 'b', parent_id => 2],
+        directory       => [dir_id => 4, dirname => 'c', parent_id => 3],
+        storage_object  => [so_id => 1, ext_id => $key1, ext_id_basename => basename($key1), dir_id => 4],
+        directory       => [dir_id => 5, dirname => 'd', parent_id => 1],
+        directory       => [dir_id => 6, dirname => 'a', parent_id => 5],
+        storage_object  => [so_id => 2, ext_id => $key2, ext_id_basename => basename($key2), dir_id => 6],
+    );
+}
+
+Test::Nebulous->setup;
+
+{
+    my $key1 = "a/b/c/foo";
+    my $key2 = "a/b/c/d/foo";
+    $neb->create_object($key1);
+    $neb->create_object($key2);
+
+    expected_dataset_ok(
+        directory       => [dir_id => 1, dirname => '/', parent_id => 1],
+        directory       => [dir_id => 2, dirname => 'a', parent_id => 1],
+        directory       => [dir_id => 3, dirname => 'b', parent_id => 2],
+        directory       => [dir_id => 4, dirname => 'c', parent_id => 3],
+        storage_object  => [so_id => 1, ext_id => $key1, ext_id_basename => basename($key1), dir_id => 4],
+        directory       => [dir_id => 5, dirname => 'd', parent_id => 4],
+        storage_object  => [so_id => 2, ext_id => $key2, ext_id_basename => basename($key2), dir_id => 5],
+    );
+}
+
 Test::Nebulous->setup;
 
Index: trunk/Nebulous-Server/t/08_server_delete_instance.t
===================================================================
--- trunk/Nebulous-Server/t/08_server_delete_instance.t	(revision 23699)
+++ trunk/Nebulous-Server/t/08_server_delete_instance.t	(revision 23932)
@@ -3,5 +3,5 @@
 # Copryight (C) 2004-2005  Joshua Hoblitt
 #
-# $Id: 08_server_delete_instance.t,v 1.12 2008-12-14 22:54:25 eugene Exp $
+# $Id: 08_server_delete_instance.t,v 1.10.22.1 2008-12-14 22:52:37 eugene Exp $
 
 use strict;
@@ -24,7 +24,8 @@
 
 {
-    my $uri = $neb->create_object("foo");
+    my $key = "foo";
+    my $uri = $neb->create_object($key);
 
-    ok($neb->delete_instance($uri), "delete instance");
+    ok($neb->delete_instance($key, $uri), "delete instance");
 }
 
@@ -32,17 +33,18 @@
 
 {
-    my $uri1 = $neb->create_object("foo");
-    my $uri2 = $neb->replicate_object("foo");
+    my $key = "foo";
+    my $uri1 = $neb->create_object($key);
+    my $uri2 = $neb->replicate_object($key);
 
-    ok($neb->delete_instance($uri1), "delete instance");
+    ok($neb->delete_instance($key, $uri1), "delete instance");
 
-    my $locations = $neb->find_instances("foo");
+    my $locations = $neb->find_instances($key);
 
     is($locations->[0], $uri2, "instance remains");
 
-    ok($neb->delete_instance( $uri2 ), "delete instance");
+    ok($neb->delete_instance($key, $uri2), "delete instance");
 
     eval {
-        $neb->find_instances("foo");
+        $neb->find_instances($key);
     };
     like($@, qr/is valid object key/, "storage object was deleted");
@@ -52,5 +54,8 @@
 
 eval {
-    $neb->delete_instance("file:/foo");
+    my $key = "foo";
+    my $uri1 = $neb->create_object($key);
+
+    $neb->delete_instance($key, "file:/foo");
 };
 like($@, qr/no instance is associated with uri/, "uri does not exist");
@@ -61,12 +66,15 @@
     $neb->delete_instance();
 };
-like($@, qr/1 was expected/, "no params");
+like($@, qr/2 were expected/, "no params");
 
 Test::Nebulous->setup;
 
 eval {
-    $neb->delete_instance("foo", 2);
+    my $key = "foo";
+    my $uri1 = $neb->create_object($key);
+
+    $neb->delete_instance("foo", 2, 3);
 };
-like($@, qr/1 was expected/, "too many params");
+like($@, qr/2 were expected/, "too many params");
 
 Test::Nebulous->cleanup;
Index: trunk/Nebulous-Server/t/10_server_is_valid_volume_name.t
===================================================================
--- trunk/Nebulous-Server/t/10_server_is_valid_volume_name.t	(revision 23699)
+++ trunk/Nebulous-Server/t/10_server_is_valid_volume_name.t	(revision 23932)
@@ -23,13 +23,13 @@
 Test::Nebulous->setup;
 
-ok($neb->_is_valid_volume_name('node01'), "valid volume name");
+ok($neb->_is_valid_volume_name('foo', 'node01'), "valid volume name");
 
 Test::Nebulous->setup;
 
-ok($neb->_is_valid_volume_name('node02'), "valid volume name");
+ok($neb->_is_valid_volume_name('foo', 'node02'), "valid volume name");
 
 Test::Nebulous->setup;
 
-is($neb->_is_valid_volume_name('node99'), undef, "invalid volume name");
+is($neb->_is_valid_volume_name('foo', 'node99'), undef, "invalid volume name");
 
 Test::Nebulous->cleanup;
Index: trunk/Nebulous-Server/t/12_server_find_objects.t
===================================================================
--- trunk/Nebulous-Server/t/12_server_find_objects.t	(revision 23699)
+++ trunk/Nebulous-Server/t/12_server_find_objects.t	(revision 23932)
@@ -8,5 +8,5 @@
 use warnings FATAL => qw( all );
 
-use Test::More tests => 6;
+use Test::More tests => 28;
 
 use lib qw( ./t ./lib );
@@ -25,6 +25,5 @@
 # search for a regex of '' should match nothing
 eval {
-    # key
-    my $uri = $neb->create_object("foo");
+    $neb->create_object("foo");
 
     my $keys = $neb->find_objects();
@@ -35,6 +34,5 @@
 
 {
-    # key
-    my $uri = $neb->create_object("foo");
+    $neb->create_object("foo");
 
     my $keys = $neb->find_objects("foo");
@@ -48,6 +46,6 @@
 {
     # key
-    my $uri1 = $neb->create_object("foo");
-    my $uri2 = $neb->replicate_object("foo");
+    $neb->create_object("foo");
+    $neb->replicate_object("foo");
 
     my $keys = $neb->find_objects("foo");
@@ -55,4 +53,120 @@
     is(scalar @$keys, 1, 'number of keys found');
     is($keys->[0], "foo", "key name");
+}
+
+Test::Nebulous->setup;
+
+{
+    # key
+    $neb->create_object("foo");
+    $neb->create_object("bar");
+
+    my $keys = $neb->find_objects("foo");
+
+    is(scalar @$keys, 1, 'number of keys found');
+    is($keys->[0], "foo", "key name");
+}
+
+# test recursive dir searching
+Test::Nebulous->setup;
+
+{
+    $neb->create_object("a/foo");
+
+    my $keys = $neb->find_objects("a");
+
+    is(scalar @$keys, 1, 'number of keys found');
+    is($keys->[0], "a/foo", "key name");
+}
+
+Test::Nebulous->setup;
+
+{
+    $neb->create_object("a/foo");
+    $neb->create_object("b/foo");
+
+    my $keys = $neb->find_objects("a");
+
+    is(scalar @$keys, 1, 'number of keys found');
+    is($keys->[0], "a/foo", "key name");
+}
+
+Test::Nebulous->setup;
+
+{
+    $neb->create_object("a/foo");
+    $neb->create_object("a/b/foo");
+
+    my $keys = $neb->find_objects("a");
+
+    is(scalar @$keys, 1, 'number of keys found');
+    is($keys->[0], "a/foo", "key name");
+}
+
+Test::Nebulous->setup;
+
+{
+    $neb->create_object("a/foo");
+    $neb->create_object("a/bar");
+
+    my $keys = $neb->find_objects("a");
+
+    is(scalar @$keys, 2, 'number of keys found');
+    is($keys->[0], "a/foo", "key name");
+    is($keys->[1], "a/bar", "key name");
+}
+
+Test::Nebulous->setup;
+
+{
+    $neb->create_object("a/foo");
+    $neb->create_object("bar");
+
+    my $keys = $neb->find_objects("a");
+
+    is(scalar @$keys, 1, 'number of keys found');
+    is($keys->[0], "a/foo", "key name");
+}
+
+Test::Nebulous->setup;
+
+{
+    $neb->create_object("a/foo");
+    $neb->create_object("foo");
+    $neb->create_object("bar");
+
+    my $keys = $neb->find_objects("/");
+
+    is(scalar @$keys, 2, 'number of keys found');
+    is($keys->[0], "foo", "key name");
+    is($keys->[1], "bar", "key name");
+}
+
+Test::Nebulous->setup;
+
+{
+    $neb->create_object("a/foo");
+    $neb->create_object("foo");
+    $neb->create_object("bar");
+
+    my $keys = $neb->find_objects(".");
+
+    is(scalar @$keys, 2, 'number of keys found');
+    is($keys->[0], "foo", "key name");
+    is($keys->[1], "bar", "key name");
+}
+
+Test::Nebulous->setup;
+
+{
+    $neb->create_object("a/foo");
+    $neb->create_object("foo");
+    $neb->create_object("bar");
+
+    my $keys = $neb->find_objects("..");
+
+    is(scalar @$keys, 2, 'number of keys found');
+    is($keys->[0], "foo", "key name");
+    is($keys->[1], "bar", "key name");
 }
 
Index: trunk/Nebulous-Server/t/13_server_rename_object.t
===================================================================
--- trunk/Nebulous-Server/t/13_server_rename_object.t	(revision 23699)
+++ trunk/Nebulous-Server/t/13_server_rename_object.t	(revision 23932)
@@ -8,8 +8,9 @@
 use warnings FATAL => qw( all );
 
-use Test::More tests => 6;
+use Test::More tests => 8;
 
 use lib qw( ./t ./lib );
 
+use File::Basename qw( basename );
 use Nebulous::Server;
 use Test::Nebulous;
@@ -21,18 +22,48 @@
 );
 
+use Test::DBUnit dsn => $NEB_DB, username => $NEB_USER, password => $NEB_PASS;
+
 Test::Nebulous->setup;
 
 {
+    my $key = "bar";
     my $uri = $neb->create_object("foo");
 
-    $neb->rename_object("foo", "bar");
+    $neb->rename_object("foo", $key);
 
-    eval {
-        $neb->find_objects('^foo$');
-    };
-    like($@, qr/no keys found/, "old key name");
+    expected_dataset_ok(
+        directory       => [dir_id => 1, dirname => '/', parent_id => 1],
+        storage_object  => [so_id => 1, ext_id => $key, ext_id_basename => basename($key), dir_id => 1],
+    );
+}
 
-    my $keys = $neb->find_objects('^bar$');
-    is(scalar @$keys, 1, 'number of keys found');
+Test::Nebulous->setup;
+
+{
+    my $key = "a/bar";
+    my $uri = $neb->create_object("foo");
+
+    $neb->rename_object("foo", $key);
+
+    expected_dataset_ok(
+        directory       => [dir_id => 1, dirname => '/', parent_id => 1],
+        directory       => [dir_id => 2, dirname => 'a', parent_id => 1],
+        storage_object  => [so_id => 1, ext_id => $key, ext_id_basename => basename($key), dir_id => 2],
+    );
+}
+
+Test::Nebulous->setup;
+
+{
+    my $key = "bar";
+    my $uri = $neb->create_object("a/foo");
+
+    $neb->rename_object("a/foo", $key);
+
+    expected_dataset_ok(
+        directory       => [dir_id => 1, dirname => '/', parent_id => 1],
+        directory       => [dir_id => 2, dirname => 'a', parent_id => 1],
+        storage_object  => [so_id => 1, ext_id => $key, ext_id_basename => basename($key), dir_id => 1],
+    );
 }
 
@@ -73,3 +104,19 @@
 like($@, qr/2 were expected/, "too many params");
 
+# test attempting to rename a key in such a way to cause the distributed
+# storage db to change
+# this must be the last test as we're messing with the $neb object
+eval {
+    $neb->config->add_db(
+        dbindex     => 1,
+        dsn         => $NEB_DB,
+        dbuser      => $NEB_USER,
+        dbpasswd    => $NEB_PASS,
+    );
+
+    $neb->create_object("a/foo");
+    $neb->rename_object("a/foo", "g/bar");
+};
+like($@, qr/rename objects across distributed database boundaries/, "rename between databases");
+
 Test::Nebulous->cleanup;
Index: trunk/Nebulous-Server/t/16_server_swap_objects.t
===================================================================
--- trunk/Nebulous-Server/t/16_server_swap_objects.t	(revision 23699)
+++ trunk/Nebulous-Server/t/16_server_swap_objects.t	(revision 23932)
@@ -8,8 +8,9 @@
 use warnings FATAL => qw( all );
 
-use Test::More tests => 8;
+use Test::More tests => 13;
 
 use lib qw( ./t ./lib );
 
+use File::Basename qw( basename );
 use Nebulous::Server;
 use Test::Nebulous;
@@ -21,14 +22,48 @@
 );
 
+use Test::DBUnit dsn => $NEB_DB, username => $NEB_USER, password => $NEB_PASS;
+
 Test::Nebulous->setup;
 
 {
-    my $uri1 = $neb->create_object("foo1");
-    my $uri2 = $neb->create_object("foo2");
+    my $key1 = "foo1";
+    my $key2 = "foo2";
+    my $uri1 = $neb->create_object($key1);
+    my $uri2 = $neb->create_object($key2);
 
-    ok($neb->swap_objects("foo1", "foo2"), "swap succeeded");
+    ok($neb->swap_objects($key1, $key2), "swap succeeded");
 
-    my $new_uri1 = ($neb->find_instances("foo1"))->[0];
-    my $new_uri2 = ($neb->find_instances("foo2"))->[0];
+    my $new_uri1 = ($neb->find_instances($key1))->[0];
+    my $new_uri2 = ($neb->find_instances($key2))->[0];
+
+    expected_dataset_ok(
+        directory       => [dir_id => 1, dirname => '/', parent_id => 1],
+        storage_object  => [so_id => 1, ext_id => $key2, ext_id_basename => basename($key2), dir_id => 1],
+        storage_object  => [so_id => 2, ext_id => $key1, ext_id_basename => basename($key1), dir_id => 1],
+    );
+
+    is($uri1, $new_uri2, "key1 -> key2");
+    is($uri2, $new_uri1, "key2 -> key1");
+}
+
+Test::Nebulous->setup;
+
+{
+    my $key1 = "foo1";
+    my $key2 = "a/foo2";
+    my $uri1 = $neb->create_object($key1);
+    my $uri2 = $neb->create_object($key2);
+
+    ok($neb->swap_objects($key1, $key2), "swap succeeded");
+
+    my $new_uri1 = ($neb->find_instances($key1))->[0];
+    my $new_uri2 = ($neb->find_instances($key2))->[0];
+
+    expected_dataset_ok(
+        directory       => [dir_id => 1, dirname => '/', parent_id => 1],
+        directory       => [dir_id => 2, dirname => 'a', parent_id => 1],
+        storage_object  => [so_id => 1, ext_id => $key2, ext_id_basename => basename($key2), dir_id => 2],
+        storage_object  => [so_id => 2, ext_id => $key1, ext_id_basename => basename($key1), dir_id => 1],
+    );
 
     is($uri1, $new_uri2, "key1 -> key2");
Index: trunk/Nebulous-Server/t/75_parse_neb_key.t
===================================================================
--- trunk/Nebulous-Server/t/75_parse_neb_key.t	(revision 23699)
+++ trunk/Nebulous-Server/t/75_parse_neb_key.t	(revision 23932)
@@ -10,9 +10,9 @@
 use Test::More;
 
-plan tests => 80;
+plan tests => 99;
 
 use lib qw( ./t ./lib );
 
-use Nebulous::Keys qw( parse_neb_key );
+use Nebulous::Key qw( parse_neb_key );
 use Test::Nebulous;
 
@@ -224,4 +224,53 @@
 }
 
+# root volume references
+{
+    my $key = parse_neb_key('neb:///');
+
+    is($key->path, '', 'path');
+    is($key->volume, undef, 'volume name');
+    is($key->soft_volume, undef, 'soft volume name');
+}
+
+{
+    my $key = parse_neb_key('neb:///.');
+
+    is($key->path, '', 'path');
+    is($key->volume, undef, 'volume name');
+    is($key->soft_volume, undef, 'soft volume name');
+}
+
+{
+    my $key = parse_neb_key('neb:///..');
+
+    is($key->path, '', 'path');
+    is($key->volume, undef, 'volume name');
+    is($key->soft_volume, undef, 'soft volume name');
+}
+
+{
+    my $key = parse_neb_key('/');
+
+    is($key->path, '', 'path');
+    is($key->volume, undef, 'volume name');
+    is($key->soft_volume, undef, 'soft volume name');
+}
+
+{
+    my $key = parse_neb_key('.');
+
+    is($key->path, '', 'path');
+    is($key->volume, undef, 'volume name');
+    is($key->soft_volume, undef, 'soft volume name');
+}
+
+{
+    my $key = parse_neb_key('..');
+
+    is($key->path, '', 'path');
+    is($key->volume, undef, 'volume name');
+    is($key->soft_volume, undef, 'soft volume name');
+}
+
 # key w/ whitespace
 eval {
@@ -267,2 +316,8 @@
 };
 like( $@, qr/requires a leading slash/, "leading slash" );
+
+# URI w/ volume but w/o path
+eval {
+    my $key = parse_neb_key('neb://foo');
+};
+like( $@, qr/requires a path/, "no path" );