Merge lp:~ubuntuone-control-tower/ubuntuone-client/generations into lp:ubuntuone-client

[Facundo Batista]

Merge a lot of 'generations' changes into trunk.

Merge a lot of 'generations' changes into trunk.

This branch is *big*, but note we've been reviewing and testing the different branches when merged into it.

[Lucio Torre (lucio.torre)]

this is correct.

review: Approve

[John Lenton (chipaca)]

This breaks trunk, and that's OK.
<cue lumberjack sketch>

review: Approve

=== modified file 'contrib/testing/testcase.py'
--- contrib/testing/testcase.py	2010-07-07 18:51:07 +0000
+++ contrib/testing/testcase.py	2010-07-22 13:39:48 +0000
@@ -201,8 +201,13 @@
         """Stub implementation."""
 
     def query_volumes(self):
-        """Stub implementation"""
-
+        """Stub implementation."""
+
+    def get_delta(self, volume_id, generation):
+        """Stub implementation."""
+
+    def rescan_from_scratch(self, volume_id):
+        """Stub implementation."""
 
 class FakeMain(main.Main):
     """ A fake Main class to setup the tests """
@@ -558,9 +563,6 @@
         """Add udf to the udfs dict."""
         self.udfs[udf.id] = udf
 
-    def handle_SYS_ROOT_RECEIVED(self, root_id):
-        """Do nothing."""
-
     def share_deleted(self, _):
         """Do nothing."""
 
@@ -683,7 +685,7 @@
 class DummyClass(object):
     """Dummy class, does nothing."""
 
-    def __getattribute__(self, name):
+    def __getattr__(self, name):
         """Any attribute is a no-op."""
         return lambda *args, **kwargs: None
 

=== modified file 'tests/syncdaemon/test_action_queue.py'
--- tests/syncdaemon/test_action_queue.py	2010-07-19 20:15:06 +0000
+++ tests/syncdaemon/test_action_queue.py	2010-07-22 13:39:48 +0000
@@ -43,18 +43,17 @@
 from contrib.mocker import Mocker
 
 from ubuntuone.storageprotocol import (
-    client, errors, protocol_pb2, volumes, request
+    client, errors, protocol_pb2, request
 )
 from ubuntuone.syncdaemon import states
 from ubuntuone.syncdaemon.action_queue import (
     ActionQueue, ActionQueueCommand, ChangePublicAccess, CreateUDF,
     DeleteVolume, Download, GetContentMixin, ListDir, ListVolumes,
     NoisyRequestQueue, RequestQueue, UploadProgressWrapper, Upload,
-    CreateShare, GetPublicFiles, VolumesQuery, GetDelta,
+    CreateShare, GetPublicFiles, GetDelta,
     TRANSFER_PROGRESS_THRESHOLD, Unlink, Move, MakeFile, MakeDir,
 )
 from ubuntuone.syncdaemon.event_queue import EventQueue, EVENTS
-from ubuntuone.syncdaemon.volume_manager import UDF
 
 
 PATH = u'~/Documents/pdfs/moño/'
@@ -789,8 +788,6 @@
         # callbacks are connected
         # pylint: disable-msg=W0212
         aq = self.action_queue
-        self.assertEqual(aq.client._node_state_callback,
-                         aq._node_state_callback)
         self.assertEqual(aq.client._share_change_callback,
                          aq._share_change_callback)
         self.assertEqual(aq.client._share_answer_callback,
@@ -1279,60 +1276,6 @@
         self.assertTrue(res is None)
 
 
-class VolumesQueryTestCase(ConnectedBaseTestCase):
-    """Test for VolumesQuery ActionQueueCommand."""
-
-    def setUp(self):
-        """Init."""
-        res = super(VolumesQueryTestCase, self).setUp()
-
-        request_queue = RequestQueue(name='fu', action_queue=self.action_queue)
-        self.command = VolumesQuery(request_queue)
-
-        return res
-
-    def test_is_action_queue_command(self):
-        """Test proper inheritance."""
-        self.assertTrue(isinstance(self.command, ActionQueueCommand))
-        self.assertTrue(isinstance(self.command, ListVolumes))
-
-    def test_run_returns_a_deferred(self):
-        """Test a deferred is returned."""
-        res = self.command._run()
-        self.assertTrue(isinstance(res, defer.Deferred), 'deferred returned')
-
-    def test_run_calls_protocol(self):
-        """Test protocol's list_volumes is called."""
-        original = self.command.action_queue.client.list_volumes
-        self.called = False
-
-        def check():
-            """Take control over client's feature."""
-            self.called = True
-
-        self.command.action_queue.client.list_volumes = check
-
-        self.command._run()
-
-        self.assertTrue(self.called, 'command was called')
-
-        self.command.action_queue.client.list_volumes = original
-
-    def test_handle_success(self):
-        """Test AQ_LIST_VOLUMES is pushed on success."""
-        request = client.ListVolumes(self.action_queue.client)
-        request.volumes = [FakedVolume(), FakedVolume()]
-        res = self.command.handle_success(success=request)
-        self.assertEqual(request.volumes, res)
-
-    def test_handle_failure(self):
-        """Test AQ_LIST_VOLUMES_ERROR is pushed on failure."""
-        msg = 'Something went wrong'
-        failure = Failure(DefaultException(msg))
-        res = self.command.handle_failure(failure=failure)
-        self.assertEqual(res, failure)
-
-
 class DeleteVolumeTestCase(ConnectedBaseTestCase):
     """Test for DeleteVolume ActionQueueCommand."""
 
@@ -1419,52 +1362,6 @@
 
         BasicTestCase.tearDown(self)
 
-    @defer.inlineCallbacks
-    def test_SV_HASH_NEW_is_pushed_for_subscrined_volume(self):
-        """SV_HASH_NEW is filtered when the volume is unsubscribed."""
-        udf_id = 'udf_id'
-        udf_volume = volumes.UDFVolume(udf_id, 'udf_node', None,
-                                       10, u'~/ñoño')
-        path = self.vm._build_udf_path(udf_volume.suggested_path)
-        udf = UDF.from_udf_volume(udf_volume, path)
-        yield self.vm.add_udf(udf)
-        yield self.vm.subscribe_udf(udf_id)
-        assert self.vm.udfs[udf_id].subscribed
-        self.action_queue.event_queue.events = [] # reset events
-
-        kwargs = dict(share_id=udf_id, node_id=NODE, hash=None)
-        self.action_queue._node_state_callback(**kwargs)
-        self.assertEqual([('SV_HASH_NEW', (), kwargs)],
-                           self.action_queue.event_queue.events)
-
-    @defer.inlineCallbacks
-    def test_SV_HASH_NEW_is_filtered_for_unsubscribed_volume(self):
-        """SV_HASH_NEW is filtered when the volume is unsubscribed."""
-        # build a VM and add it an UDF with subscribed to False
-        udf_id = 'udf_id'
-        udf_volume = volumes.UDFVolume(udf_id, 'udf_node', None, 10, u'~/ñoño')
-        path = self.vm._build_udf_path(udf_volume.suggested_path)
-        udf = UDF.from_udf_volume(udf_volume, path)
-        yield self.vm.add_udf(udf)
-        yield self.vm.unsubscribe_udf(udf_id)
-        assert not self.vm.udfs[udf_id].subscribed
-        self.action_queue.event_queue.events = [] # reset events
-
-        self.action_queue._node_state_callback(share_id=udf_id,
-                                               node_id=None, hash=None)
-        self.assertEqual([], self.action_queue.event_queue.events)
-
-    def test_SV_HASH_NEW_doesnt_fail_for_non_udf(self):
-        """SV_HASH_NEW keeps working like before for non-udfs."""
-        other_id = 'not in udfs'
-        assert other_id not in self.vm.udfs
-        self.action_queue.event_queue.events = [] # reset events
-
-        kwargs = dict(share_id=other_id, node_id=NODE, hash=None)
-        self.action_queue._node_state_callback(**kwargs)
-        self.assertEqual([('SV_HASH_NEW', (), kwargs)],
-                          self.action_queue.event_queue.events)
-
 
 class ChangePublicAccessTests(ConnectedBaseTestCase):
     """Tests for the ChangePublicAccess ActionQueueCommand."""
@@ -2222,19 +2119,23 @@
 class SimpleAQTestCase(BasicTestCase):
     """Simple tests for AQ API."""
 
-    def test_aq_server_rescan(self):
-        """Check the API of AQ.server_rescan."""
+    def test_aq_query_volumes(self):
+        """Check the API of AQ.query_volumes."""
         self.main.start()
         d = defer.Deferred()
-        def get_root(mdid):
-            """Fake get_root."""
-            d.callback(mdid)
+        def list_volumes():
+            """Fake list_volumes."""
+            result = DummyClass()
+            result.volumes = ['foo', 'bar']
+            return defer.succeed(result)
 
         self.action_queue.client = DummyClass()
-        self.action_queue.get_root = get_root
-        self.action_queue.server_rescan('foo', lambda: list())
+        self.action_queue.client.list_volumes = list_volumes
+        d = self.action_queue.query_volumes()
         def check(result):
-            self.assertEqual('foo', result)
+            self.assertIn('foo', result)
+            self.assertIn('bar', result)
+            return result
         d.addCallback(check)
         return d
 
@@ -2783,81 +2684,6 @@
         # assert internal deferred was fired
         self.assertTrue(self.action_queue.deferred.called)
 
-    @defer.inlineCallbacks
-    def test_server_rescan_as_a_whole(self):
-        """Test error handling after server_rescan with no error."""
-
-        def faked_get_root(marker):
-            """Fake the action_queue.get_root."""
-            root_id=object()
-            self.action_queue.event_queue.push('SYS_ROOT_RECEIVED',
-                                               root_id=root_id)
-            return root_id
-
-        self.patch(self.action_queue, 'get_root', faked_get_root)
-
-        self.action_queue.client.query = \
-            self.succeed_please(result=self.action_queue.client)
-        yield self.action_queue.server_rescan(root_mdid=object(),
-                                              data_gen=list)
-        event = ('SYS_SERVER_RESCAN_DONE', (), {})
-        self.assertEqual(event, self.action_queue.event_queue.events[-1])
-
-        # assert internal deferred wasn't fired
-        self.assertFalse(self.action_queue.deferred.called)
-
-    @defer.inlineCallbacks
-    def test_server_rescan_when_get_root_fails(self):
-        """Test error handling after server_rescan when get_root fails."""
-
-        msg = protocol_pb2.Message()
-        msg.type = protocol_pb2.Message.ERROR
-        msg.error.type = protocol_pb2.Error.PROTOCOL_ERROR
-        msg.error.comment = 'get_root failed'
-        exc = errors.StorageRequestError(request=None, message=msg)
-        self.patch(self.action_queue, 'get_root', self.fail_please(exc))
-
-        self.action_queue.client.query = self.fail_please(
-                                                        NotImplementedError())
-
-        yield self.action_queue.server_rescan(root_mdid=object(),
-                                              data_gen=list)
-
-        event = ('SYS_SERVER_RESCAN_DONE', (), {})
-        self.assertNotIn(event, self.action_queue.event_queue.events)
-
-        event = ('SYS_SERVER_ERROR', (), {'error': str(exc)})
-        self.assertEqual(event, self.action_queue.event_queue.events[-1])
-
-        # assert internal deferred wasn't fired
-        self.assertFalse(self.action_queue.deferred.called)
-
-    @defer.inlineCallbacks
-    def test_server_rescan_when_query_fails(self):
-        """Test error handling after server_rescan when query fails."""
-
-        self.patch(self.action_queue, 'get_root',
-                   self.succeed_please(result=object()))
-
-        msg = protocol_pb2.Message()
-        msg.type = protocol_pb2.Message.ERROR
-        msg.error.type = protocol_pb2.Error.PROTOCOL_ERROR
-        msg.error.comment = 'query failed'
-        exc = errors.StorageRequestError(request=None, message=msg)
-        self.action_queue.client.query = self.fail_please(exc)
-
-        yield self.action_queue.server_rescan(root_mdid=object(),
-                                              data_gen=list)
-
-        event = ('SYS_SERVER_RESCAN_DONE', (), {})
-        self.assertNotIn(event, self.action_queue.event_queue.events)
-
-        event = ('SYS_SERVER_ERROR', (), {'error': str(exc)})
-        self.assertEqual(event, self.action_queue.event_queue.events[-1])
-
-        # assert internal deferred wasn't fired
-        self.assertFalse(self.action_queue.deferred.called)
-
 
 class GetDeltaTestCase(ConnectedBaseTestCase):
     """Test for GetDelta ActionQueueCommand."""
@@ -2912,7 +2738,8 @@
 
         # check for successful event
         received = self.action_queue.event_queue.events[0]
-        delta_info = dict(delta_content=['foo', 'bar'], end_generation=76,
+        delta_info = dict(volume_id=VOLUME, delta_content=['foo', 'bar'],
+                          end_generation=76,
                           full=True, free_bytes=1231234)
         self.assertEqual(received, ('AQ_DELTA_OK', (), delta_info))
 

=== modified file 'tests/syncdaemon/test_dbus.py'
--- tests/syncdaemon/test_dbus.py	2010-07-08 21:52:23 +0000
+++ tests/syncdaemon/test_dbus.py	2010-07-22 13:39:48 +0000
@@ -1508,8 +1508,8 @@
             self.assertEquals('root_id', root_id)
             self.assertEquals('another_root_id', new_root_id)
         d.addCallback(check)
-        self.event_q.push('SYS_ROOT_RECEIVED', 'root_id')
-        self.event_q.push('SYS_ROOT_RECEIVED', 'another_root_id')
+        self.main.vm._got_root('root_id')
+        self.main.vm._got_root('another_root_id')
         return d
 
     def test_public_files_list(self):

=== modified file 'tests/syncdaemon/test_sync.py'
--- tests/syncdaemon/test_sync.py	2010-07-20 16:17:58 +0000
+++ tests/syncdaemon/test_sync.py	2010-07-22 13:39:48 +0000
@@ -19,6 +19,7 @@
 
 from __future__ import with_statement
 
+import copy
 import logging
 import os
 import shutil
@@ -43,6 +44,8 @@
 from ubuntuone.syncdaemon.volume_manager import Share
 from ubuntuone.syncdaemon.event_queue import EventQueue
 from ubuntuone.storageprotocol.request import ROOT
+from ubuntuone.storageprotocol import delta
+from ubuntuone.syncdaemon.marker import MDMarker
 
 DBusInterface.test = True
 
@@ -312,7 +315,7 @@
 
         # send the event and check
         mdobj = fsm.get_by_mdid(mdid)
-        sync.handle_SV_HASH_NEW(mdobj.share_id, mdobj.node_id, '') # no content
+        sync._handle_SV_HASH_NEW(mdobj.share_id, mdobj.node_id, '') # no content
         self.assertTrue(self.called)
 
 
@@ -464,81 +467,81 @@
     def test_handle_AQ_UNLINK_OK(self):
         """Test that AQ_UNLINK_OK calls the generation handler."""
         called = []
-        self.patch(SyncStateMachineRunner, 'check_new_volume_generation',
+        self.patch(SyncStateMachineRunner, 'update_generation',
                    lambda s, *a: called.append(a))
 
         d = dict(share_id='volume_id', node_id='node_id', parent_id='parent',
                  new_generation=77)
         self.sync.handle_AQ_UNLINK_OK(**d)
-        self.assertEqual(called, [('volume_id', 77)])
+        self.assertEqual(called, [('volume_id', "node_id", 77)])
 
     def test_handle_AQ_MOVE_OK(self):
         """Test that AQ_MOVE_OK calls the generation handler."""
         called = []
-        self.patch(SyncStateMachineRunner, 'check_new_volume_generation',
+        self.patch(SyncStateMachineRunner, 'update_generation',
                    lambda s, *a: called.append(a))
 
         d = dict(share_id='volume_id', node_id='node_id', new_generation=32)
         self.sync.handle_AQ_MOVE_OK(**d)
-        self.assertEqual(called, [('volume_id', 32)])
+        self.assertEqual(called, [('volume_id', "node_id", 32)])
 
     def test_handle_AQ_UPLOAD_FINISHED(self):
         """Test that AQ_UPLOAD_FINISHED calls the generation handler."""
         called = []
-        self.patch(SyncStateMachineRunner, 'check_new_volume_generation',
+        self.patch(SyncStateMachineRunner, 'update_generation',
                    lambda s, *a: called.append(a))
 
         d = dict(share_id='volume_id', node_id='node_id',
                  hash='hash', new_generation=15)
         self.sync.handle_AQ_UPLOAD_FINISHED(**d)
-        self.assertEqual(called, [('volume_id', 15)])
+        self.assertEqual(called, [('volume_id', "node_id", 15)])
 
     def test_handle_AQ_FILE_NEW_OK(self):
         """Test that AQ_FILE_NEW_OK calls the generation handler."""
         called = []
-        self.patch(SyncStateMachineRunner, 'check_new_volume_generation',
+        self.patch(SyncStateMachineRunner, 'update_generation',
                    lambda s, *a: called.append(a))
 
         d = dict(marker='mdid', new_id='new_id', new_generation=12,
                  volume_id=ROOT)
         self.sync.handle_AQ_FILE_NEW_OK(**d)
-        self.assertEqual(called, [(ROOT, 12)])
+        self.assertEqual(called, [(ROOT, "new_id", 12)])
 
     def test_handle_AQ_DIR_NEW_OK(self):
         """Test that AQ_DIR_NEW_OK calls the generation handler."""
         called = []
-        self.patch(SyncStateMachineRunner, 'check_new_volume_generation',
+        self.patch(SyncStateMachineRunner, 'update_generation',
                    lambda s, *a: called.append(a))
 
         d = dict(marker='mdid', new_id='new_id', new_generation=17,
                  volume_id=ROOT)
         self.sync.handle_AQ_DIR_NEW_OK(**d)
-        self.assertEqual(called, [(ROOT, 17)])
+        self.assertEqual(called, [(ROOT, "new_id", 17)])
 
     def test_checknewvol_no_volume(self):
         """Log warning if volume does not exist."""
         not_existant_vol = str(uuid.uuid4())
-        self.ssmr.check_new_volume_generation(not_existant_vol, 77)
+        self.ssmr.update_generation(not_existant_vol, "node_id", 77)
         self.assertTrue(self.handler.check_warning('Volume not found'))
 
     def test_checknewvol_smaller_gen(self):
         """Only log debug if new generation smaller than current."""
         self.vm.update_generation(ROOT, 15)
-        self.ssmr.check_new_volume_generation(ROOT, 14)
+        self.ssmr.update_generation(ROOT, "node_id", 14)
         self.assertTrue(self.handler.check_info(
                         'Got smaller or equal generation'))
 
     def test_checknewvol_same_gen(self):
         """Only log debug if new generation equal than current."""
         self.vm.update_generation(ROOT, 15)
-        self.ssmr.check_new_volume_generation(ROOT, 15)
+        self.ssmr.update_generation(ROOT, "node_id", 15)
         self.assertTrue(self.handler.check_info(
                         'Got smaller or equal generation'))
 
     def test_checknewvol_gen_current_plus_one(self):
         """Set new volume generation if current plus one."""
         self.vm.update_generation(ROOT, 15)
-        self.ssmr.check_new_volume_generation(ROOT, 16)
+        self.ssmr.update_generation(ROOT, "node_id", 16)
         self.assertEqual(self.vm.get_volume(ROOT).generation, 16)
         self.assertTrue(self.handler.check_info('Updating current generation'))
 
@@ -550,7 +553,7 @@
         self.vm.update_generation(ROOT, 15)
 
         # call the method
-        self.ssmr.check_new_volume_generation(ROOT, 17)
+        self.ssmr.update_generation(ROOT, "node_id", 17)
 
         # check that generation didn't change, we asked for delta, and logged
         self.assertEqual(self.vm.get_volume(ROOT).generation, 15)
@@ -560,13 +563,315 @@
     def test_checknewvol_new_gen_is_None(self):
         """Log warning if volume does not exist."""
         self.vm.update_generation(ROOT, 1)
-        self.ssmr.check_new_volume_generation(ROOT, None)
+        self.ssmr.update_generation(ROOT, "node_id", None)
         self.assertTrue(self.handler.check_debug(
                         'Client not ready for generations'))
 
     def test_checknewvol_volume_gen_is_None(self):
         """Log warning if volume does not exist."""
         assert self.vm.get_volume(ROOT).generation is None
-        self.ssmr.check_new_volume_generation(ROOT, 15)
+        self.ssmr.update_generation(ROOT, "node_id", 15)
         self.assertTrue(self.handler.check_debug(
                         'Client not ready for generations'))
+
+    def test_check_generation_on_node_set(self):
+        """Check that we update the generation of the node."""
+        # create the fake file
+        self.main.vm._got_root("parent_id")
+        self.sync._handle_SV_FILE_NEW(ROOT, "node_id", "parent_id", "file")
+
+        # update generation
+        self.ssmr.update_generation(ROOT, "node_id", 15)
+
+        # test
+        node = self.main.fs.get_by_node_id(ROOT, "node_id")
+        self.assertEqual(node.generation, 15)
+
+    def test_check_generation_on_node_set_wont_fail(self):
+        """Check that if there is no node we dont fail."""
+        # update generation
+        self.ssmr.update_generation(ROOT, "node_id", 15)
+
+    def test_save_generation_after_seting_node_id(self):
+        """Test that we call update_generation after the ssmr handler."""
+        root_id = uuid.uuid4()
+        self.main.vm._got_root(root_id)
+        mdobj = self.main.fs.get_by_node_id(ROOT, root_id)
+        path = os.path.join(
+            self.main.fs.get_abspath(ROOT, mdobj.path), "file")
+        self.main.fs.create(path=path, share_id=ROOT, is_dir=False)
+        node = self.main.fs.get_by_path(path)
+        d = dict(marker=MDMarker(node.mdid),
+                 new_id='new_id', new_generation=12,
+                 volume_id=ROOT)
+        self.sync.handle_AQ_FILE_NEW_OK(**d)
+
+        # test
+        node = self.main.fs.get_by_node_id(ROOT, "new_id")
+        self.assertEqual(node.generation, 12)
+
+class TestHandleAqDeltaOk(BaseSync):
+    """Sync.handle_AQ_DELTA_OK handles the recepcion of a new delta and applies
+    all the changes that came from it."""
+
+    def setUp(self):
+        """Do the setUp."""
+
+        super(TestHandleAqDeltaOk, self).setUp()
+        self.sync = Sync(main=self.main)
+        root_id = uuid.uuid4()
+        self.main.vm._got_root(root_id)
+
+        self.filetxtdelta = delta.FileInfoDelta(
+            generation=5, is_live=True, file_type=delta.FILE,
+            parent_id=root_id, share_id=ROOT, node_id=uuid.uuid4(),
+            name="file.txt", is_public=False, content_hash="hash",
+            crc32=1, size=10, last_modified=0)
+
+        self.dirdelta = delta.FileInfoDelta(
+            generation=6, is_live=True, file_type=delta.DIRECTORY,
+            parent_id=root_id, share_id=ROOT, node_id=uuid.uuid4(),
+            name="directory", is_public=False, content_hash="hash",
+            crc32=1, size=10, last_modified=0)
+
+    def create_filetxt(self):
+        """Create a file based on self.filetxtdelta."""
+
+        dt = self.filetxtdelta
+        mdobj = self.main.fs.get_by_node_id(dt.share_id, dt.parent_id)
+        path = os.path.join(
+            self.main.fs.get_abspath(dt.share_id, mdobj.path), dt.name)
+        self.main.fs.create(
+            path=path, share_id=dt.share_id, node_id=dt.node_id,
+            is_dir=False)
+        node = self.main.fs.get_by_node_id(dt.share_id, dt.node_id)
+        self.main.fs.set_by_mdid(node.mdid, generation=dt.generation)
+
+    def create_dir(self):
+        """Create a directory based on self.dirdelta."""
+
+        dt = self.dirdelta
+        mdobj = self.main.fs.get_by_node_id(dt.share_id, dt.parent_id)
+        path = os.path.join(
+            self.main.fs.get_abspath(dt.share_id, mdobj.path), dt.name)
+        self.main.fs.create(
+            path=path, share_id=dt.share_id, node_id=dt.node_id,
+            is_dir=True)
+        node = self.main.fs.get_by_node_id(dt.share_id, dt.node_id)
+        self.main.fs.set_by_mdid(node.mdid, generation=dt.generation)
+
+    def test_not_full(self):
+        """If we dont have a full delta, we need to ask for another one."""
+        sync = Sync(main=self.main)
+        called = []
+        self.main.action_q.get_delta = lambda *a: called.append(a)
+
+        kwargs = dict(volume_id=ROOT, delta_content=[], end_generation=11,
+                      full=False, free_bytes=0)
+        sync.handle_AQ_DELTA_OK(**kwargs)
+
+        self.assertEqual(called, [(ROOT, 11)])
+
+    def test_free_bytes_set(self):
+        """The volume gets the free bytes set."""
+        sync = Sync(main=self.main)
+
+        kwargs = dict(volume_id=ROOT, delta_content=[], end_generation=11,
+                      full=True, free_bytes=10)
+        sync.handle_AQ_DELTA_OK(**kwargs)
+
+        self.assertEqual(self.main.vm.get_volume(ROOT).free_bytes, 10)
+
+    def test_end_generation_set(self):
+        """The volume gets the end generation set."""
+        sync = Sync(main=self.main)
+
+        kwargs = dict(volume_id=ROOT, delta_content=[], end_generation=11,
+                      full=True, free_bytes=10)
+        sync.handle_AQ_DELTA_OK(**kwargs)
+
+        self.assertEqual(self.main.vm.get_volume(ROOT).generation, 11)
+
+    def test_node_generation_older_skip(self):
+        """The node does not get the new generation set."""
+        self.create_filetxt()
+
+        dt2 = copy.copy(self.filetxtdelta)
+        dt2.generation = self.filetxtdelta.generation - 1
+        kwargs = dict(volume_id=ROOT, delta_content=[dt2], end_generation=11,
+                      full=True, free_bytes=10)
+        self.sync.handle_AQ_DELTA_OK(**kwargs)
+
+        node = self.main.fs.get_by_node_id(ROOT, self.filetxtdelta.node_id)
+        self.assertEqual(node.generation, self.filetxtdelta.generation)
+
+    def test_new_file(self):
+        """Make sure a live file in the delta is in fs after executed."""
+        deltas = [ self.filetxtdelta ]
+
+        kwargs = dict(volume_id=ROOT, delta_content=deltas, end_generation=11,
+                      full=True, free_bytes=10)
+        self.sync.handle_AQ_DELTA_OK(**kwargs)
+
+        # check that the file is created
+        node = self.main.fs.get_by_node_id(ROOT, self.filetxtdelta.node_id)
+        self.assertEqual(node.path, self.filetxtdelta.name)
+        self.assertEqual(node.is_dir, False)
+        self.assertEqual(node.generation, self.filetxtdelta.generation)
+
+    def test_existing_file_still_there(self):
+        """A file will still exist after a delta arrives."""
+        self.create_filetxt()
+
+        # send a new delta
+        dt2 = copy.copy(self.filetxtdelta)
+        dt2.generation = 8
+        kwargs = dict(volume_id=ROOT, delta_content=[dt2], end_generation=11,
+                      full=True, free_bytes=10)
+        self.sync.handle_AQ_DELTA_OK(**kwargs)
+
+        # check that the file is created
+        node = self.main.fs.get_by_node_id(ROOT, self.filetxtdelta.node_id)
+        self.assertEqual(node.generation, dt2.generation)
+
+    def test_existing_file_dead(self):
+        """The handler for SV_FILE_DELETED is called"""
+        # send a new delta
+        dt2 = copy.copy(self.filetxtdelta)
+        dt2.is_live = False
+        kwargs = dict(volume_id=ROOT, delta_content=[dt2], end_generation=11,
+                      full=True, free_bytes=10)
+        called = []
+        self.sync._handle_SV_FILE_DELETED = \
+                   lambda *args, **kwargs: called.append((args, kwargs))
+        self.sync.handle_AQ_DELTA_OK(**kwargs)
+
+        # check that the handler is created
+        self.assertTrue(called)
+
+    def test_new_dir(self):
+        """Make sure a live dir in the delta is in fs after executed."""
+
+        deltas = [ self.dirdelta ]
+
+        kwargs = dict(volume_id=ROOT, delta_content=deltas, end_generation=11,
+                      full=True, free_bytes=10)
+        self.sync.handle_AQ_DELTA_OK(**kwargs)
+
+        # check that the dir is created
+        node = self.main.fs.get_by_node_id(ROOT, self.dirdelta.node_id)
+        self.assertEqual(node.path, self.dirdelta.name)
+        self.assertEqual(node.is_dir, True)
+        self.assertEqual(node.generation, self.dirdelta.generation)
+
+    def test_sv_hash_new_called_for_file(self):
+        """The handler for SV_HASH_NEW is called"""
+        self.create_filetxt()
+
+        # send a new delta
+        dt2 = copy.copy(self.filetxtdelta)
+        dt2.generation = self.filetxtdelta.generation + 1
+        kwargs = dict(volume_id=ROOT, delta_content=[dt2], end_generation=11,
+                      full=True, free_bytes=10)
+        called = []
+        self.sync._handle_SV_HASH_NEW = \
+                   lambda *args, **kwargs: called.append((args, kwargs))
+        self.sync.handle_AQ_DELTA_OK(**kwargs)
+
+        # check that the handler is created
+        self.assertTrue(called)
+
+    def test_sv_hash_new_not_called_for_dir(self):
+        """The handler for SV_HASH_NEW is not called"""
+        self.create_dir()
+
+        # send a new delta
+        dt2 = copy.copy(self.dirdelta)
+        dt2.generation = self.dirdelta.generation + 1
+        kwargs = dict(volume_id=ROOT, delta_content=[dt2], end_generation=11,
+                      full=True, free_bytes=10)
+        called = []
+        self.sync._handle_SV_HASH_NEW = \
+                   lambda *args, **kwargs: called.append((args, kwargs))
+        self.sync.handle_AQ_DELTA_OK(**kwargs)
+
+        # check that the handler is created
+        self.assertFalse(called)
+
+    def test_sv_moved_called(self):
+        """The handler for SV_MOVED is called"""
+        self.create_dir()
+        self.create_filetxt()
+
+        # send a new delta
+        dt2 = copy.copy(self.filetxtdelta)
+        dt2.generation = self.dirdelta.generation + 1
+        dt2.parent_id = self.dirdelta.node_id
+        kwargs = dict(volume_id=ROOT, delta_content=[dt2], end_generation=11,
+                      full=True, free_bytes=10)
+        called = []
+        self.sync._handle_SV_MOVED = \
+                   lambda *args, **kwargs: called.append((args, kwargs))
+        self.sync.handle_AQ_DELTA_OK(**kwargs)
+
+        # check that the handler is created
+        self.assertTrue(called)
+
+    def test_sv_moved_called_name(self):
+        """The handler for SV_MOVED is called"""
+        self.create_dir()
+        self.create_filetxt()
+
+        # send a new delta
+        dt2 = copy.copy(self.filetxtdelta)
+        dt2.generation = self.dirdelta.generation + 1
+        dt2.name = "newname"
+        kwargs = dict(volume_id=ROOT, delta_content=[dt2], end_generation=11,
+                      full=True, free_bytes=10)
+        called = []
+        self.sync._handle_SV_MOVED = \
+                   lambda *args, **kwargs: called.append((args, kwargs))
+        self.sync.handle_AQ_DELTA_OK(**kwargs)
+
+        # check that the handler is created
+        self.assertTrue(called)
+
+    def test_sv_moved_not_called(self):
+        """The handler for SV_MOVED is not called"""
+        self.create_dir()
+        self.create_filetxt()
+
+        # send a new delta
+        dt2 = copy.copy(self.filetxtdelta)
+        dt2.generation = self.dirdelta.generation + 1
+        kwargs = dict(volume_id=ROOT, delta_content=[dt2], end_generation=11,
+                      full=True, free_bytes=10)
+        called = []
+        self.sync._handle_SV_MOVED = \
+                   lambda *args, **kwargs: called.append((args, kwargs))
+        self.sync.handle_AQ_DELTA_OK(**kwargs)
+
+        # check that the handler is created
+        self.assertFalse(called)
+
+    def test_exception_logged(self):
+        """We call self.logger.exception on error."""
+        # send a new delta
+        dt2 = copy.copy(self.filetxtdelta)
+        dt2.is_live = False
+        kwargs = dict(volume_id=ROOT, delta_content=[dt2], end_generation=11,
+                      full=True, free_bytes=10)
+        self.sync._handle_SV_FILE_DELETED = \
+                   lambda *args, **kwargs: 1/0
+        handler = testcase.MementoHandler()
+        handler.setLevel(logging.ERROR)
+        self.sync.logger.addHandler(handler)
+
+        self.sync.handle_AQ_DELTA_OK(**kwargs)
+
+        # check log
+        self.assertEqual(len(handler.records), 1)
+        log_msg = handler.records[0].message
+        self.assertTrue("can't be applied." in log_msg)
+        self.sync.logger.removeHandler(handler)
+        self.handler.records = []

=== modified file 'tests/syncdaemon/test_vm.py'
--- tests/syncdaemon/test_vm.py	2010-07-12 15:47:38 +0000
+++ tests/syncdaemon/test_vm.py	2010-07-22 13:39:48 +0000
@@ -32,6 +32,7 @@
     FakeMain,
     BaseTwistedTestCase,
     environ,
+    MementoHandler,
 )
 from ubuntuone.syncdaemon import config
 from ubuntuone.syncdaemon.volume_manager import (
@@ -115,22 +116,27 @@
 class VolumeManagerTests(BaseVolumeManagerTests):
     """ Tests for Volume Manager internal API. """
 
-    def test_handle_SYS_ROOT_RECEIVED(self):
-        """Check that list_shares is called in handle_SYS_ROOT_RECEIVED """
-        d = defer.Deferred()
-        # helper method, pylint: disable-msg=C0111
-        def list_shares():
-            mdobj = self.main.fs.get_by_path(self.root_dir)
-            self.assertEquals('root_uuid', mdobj.node_id)
-            d.callback(mdobj)
-        self.main.action_q.list_shares = list_shares
-        self.vm.handle_SYS_ROOT_RECEIVED('root_uuid')
-        return d
+    @defer.inlineCallbacks
+    def test__got_root_ok(self):
+        """Test _got_root method."""
+        d = defer.Deferred()
+        self._listen_for('SYS_ROOT_RECEIVED', d.callback)
+        self.vm._got_root('root_uuid')
+        yield d
+
+    @defer.inlineCallbacks
+    def test__got_root_mismatch(self):
+        """Test for _got_root with different root node_id."""
+        self.vm._got_root('root_uuid')
+        d = defer.Deferred()
+        self._listen_for('SYS_ROOT_MISMATCH', d.callback)
+        self.vm._got_root('other_root_uuid')
+        yield d
 
     def test_add_share(self):
         """ test the add_share method. """
         # initialize the the root
-        self.vm.handle_SYS_ROOT_RECEIVED('root_uuid')
+        self.vm._got_root('root_uuid')
         share_path = os.path.join(self.shares_dir, 'fake_share')
         share = Share(path=share_path, volume_id='share_id')
         self.vm.add_share(share)
@@ -139,7 +145,7 @@
     def test_share_deleted(self):
         """ Check that a share is deleted from the share mapping. """
         # initialize the the root
-        self.vm.handle_SYS_ROOT_RECEIVED('root_uuid')
+        self.vm._got_root('root_uuid')
         share_path = os.path.join(self.shares_dir, 'fake_share')
         share = Share(path=share_path, volume_id='share_id')
         self.vm.add_share(share)
@@ -150,7 +156,7 @@
     def test_share_deleted_with_content(self):
         """ Check that a share is deleted from the share mapping. """
         # initialize the the root
-        self.vm.handle_SYS_ROOT_RECEIVED('root_uuid')
+        self.vm._got_root('root_uuid')
         share_path = os.path.join(self.shares_dir, 'fake_share')
         share = Share(path=share_path, volume_id='share_id',
                       node_id='share_node_id', access_level='Modify',
@@ -192,7 +198,7 @@
                                                      'test_username',
                                                      'visible_name', 'Modify')
         # initialize the the root
-        self.vm.handle_SYS_ROOT_RECEIVED('root_uuid')
+        self.vm._got_root('root_uuid')
         share_path = os.path.join(self.shares_dir, share_holder.share_name)
         share = Share(path=share_path, volume_id=share_holder.share_id,
                       access_level='View')
@@ -209,7 +215,7 @@
                                                    'visible_username', 'yes',
                                                    'View')
         # initialize the the root
-        self.vm.handle_SYS_ROOT_RECEIVED('root_uuid')
+        self.vm._got_root('root_uuid')
         response = ListShares(None)
         response.shares = [share_response]
         self.vm.handle_AQ_SHARES_LIST(response)
@@ -228,7 +234,7 @@
                                                      'test_username',
                                                      'visible_name', 'Modify')
         # initialize the the root
-        self.vm.handle_SYS_ROOT_RECEIVED('root_uuid')
+        self.vm._got_root('root_uuid')
         # create a share
         share_path = os.path.join(self.shares_dir, share_holder.share_name)
         share = Share(path=share_path, volume_id=str(share_holder.share_id),
@@ -265,7 +271,7 @@
                                                    'my_visible_name', 'yes',
                                                    'Modify')
         # initialize the the root
-        self.vm.handle_SYS_ROOT_RECEIVED('root_uuid')
+        self.vm._got_root('root_uuid')
         shared_dir = os.path.join(self.root_dir, 'shared_dir')
         self.main.fs.create(path=shared_dir, share_id="", is_dir=True)
         self.main.fs.set_node_id(shared_dir, shared_response.subtree)
@@ -374,7 +380,7 @@
     def test_accept_share(self):
         """ Test the accept_share method. """
         d = defer.Deferred()
-        self.vm.handle_SYS_ROOT_RECEIVED('root_uuid')
+        self.vm._got_root('root_uuid')
         share_path = os.path.join(self.shares_dir, 'fake_share')
         share = Share(path=share_path, volume_id='share_id', node_id="node_id")
         self.vm.add_share(share)
@@ -403,7 +409,7 @@
                                                    'my_visible_name', 'yes',
                                                    'Modify')
         # initialize the the root
-        self.vm.handle_SYS_ROOT_RECEIVED('root_uuid')
+        self.vm._got_root('root_uuid')
         response = ListShares(None)
         response.shares = [shared_response]
         self.vm.handle_AQ_SHARES_LIST(response)
@@ -421,7 +427,7 @@
         self.main.fs.set_node_id(path, 'node_id')
         self.main.fs.get_by_node_id("", 'node_id')
         # initialize the the root
-        self.vm.handle_SYS_ROOT_RECEIVED('root_uuid')
+        self.vm._got_root('root_uuid')
         # add the shared folder
         share = Share(path=path, volume_id='share_id', access_level='View')
         self.vm.add_shared(share)
@@ -437,7 +443,7 @@
         self.main.fs.set_node_id(path, 'node_id')
         self.main.fs.get_by_node_id("", 'node_id')
         # initialize the the root
-        self.vm.handle_SYS_ROOT_RECEIVED('root_uuid')
+        self.vm._got_root('root_uuid')
         self.assertNotIn('share_id', self.vm.shared)
         # check that a answer notify of a missing share don't blowup
         self.vm.handle_SV_SHARE_ANSWERED('share_id', 'Yes')
@@ -459,7 +465,7 @@
                                                    'visible_username', False,
                                                    'View')
         # initialize the the root
-        self.vm.handle_SYS_ROOT_RECEIVED('root_uuid')
+        self.vm._got_root('root_uuid')
         response = ListShares(None)
         response.shares = [share_response, share_response_1]
         self.vm.handle_AQ_SHARES_LIST(response)
@@ -554,14 +560,43 @@
     @defer.inlineCallbacks
     def test_root_mismatch(self):
         """Test that SYS_ROOT_MISMATCH is pushed."""
-        self.vm.handle_SYS_ROOT_RECEIVED('root_node_id')
+        self.vm._got_root('root_node_id')
         d = defer.Deferred()
         self._listen_for('SYS_ROOT_MISMATCH', d.callback)
-        self.vm.handle_SYS_ROOT_RECEIVED('root_id')
+        self.vm._got_root('root_id')
         current_id, new_id = yield d
         self.assertEquals('root_node_id', current_id)
         self.assertEquals('root_id', new_id)
 
+    @defer.inlineCallbacks
+    def test_handle_AQ_ANSWER_SHARE_OK(self):
+        """Test for handle_AQ_ANSWER_SHARE_OK."""
+        share_id = uuid.uuid4()
+        share_volume = volumes.ShareVolume(share_id, 'fake_share_uuid', None,
+                                           10, 'to_me', 'fake_share', 'username',
+                                           'visible_username', False, 'View')
+        dir_name = self.vm._build_share_path(share_volume.share_name,
+                                          share_volume.other_visible_name)
+        share_path = os.path.join(self.main.shares_dir, dir_name)
+        share = Share.from_share_volume(share_volume, share_path)
+
+        get_delta_d = defer.Deferred()
+        def fake_get_delta(volume_id, generation):
+            """A fake get_delta that check the arguments."""
+            self.assertEquals(share.volume_id, volume_id)
+            self.assertEquals(share.generation, generation)
+            get_delta_d.callback(None)
+        self.main.action_q.get_delta = fake_get_delta
+
+        self.vm.add_share(share)
+        self.vm.handle_AQ_ANSWER_SHARE_OK(share.volume_id, 'Yes')
+        yield get_delta_d
+        share = self.vm.get_volume(share.volume_id)
+        self.assertTrue(share.accepted, 'accepted != True')
+        self.assertTrue(self.main.fs.get_by_path(share.path),
+                        'No metadata for share root node.')
+        self.assertTrue(os.path.exists(share.path), 'share path missing on disk!')
+
 
 class VolumeManagerUnicodeTests(BaseVolumeManagerTests):
     """Tests for Volume Manager unicode capabilities."""
@@ -574,7 +609,7 @@
                                                    'visible', 'yes',
                                                    'View')
         # initialize the the root
-        self.vm.handle_SYS_ROOT_RECEIVED('root_uuid')
+        self.vm._got_root('root_uuid')
         response = ListShares(None)
         response.shares = [share_response]
         self.vm.handle_AQ_SHARES_LIST(response)
@@ -593,7 +628,7 @@
                                                    u'Darío Toño', 'yes',
                                                    'View')
         # initialize the the root
-        self.vm.handle_SYS_ROOT_RECEIVED('root_uuid')
+        self.vm._got_root('root_uuid')
         response = ListShares(None)
         response.shares = [share_response]
         self.vm.handle_AQ_SHARES_LIST(response)
@@ -610,7 +645,7 @@
                                                      u'año',
                                                      'test_username',
                                                      'visible', 'Modify')
-        self.vm.handle_SYS_ROOT_RECEIVED('root_uuid')
+        self.vm._got_root('root_uuid')
         self.vm.handle_SV_SHARE_CHANGED(info=share_holder)
         shouldbe_dir = os.path.join(self.shares_dir,
                                     u"año".encode("utf8") + " from visible")
@@ -622,7 +657,7 @@
                                                      'share',
                                                      'test_username',
                                                      u'Ramón', 'Modify')
-        self.vm.handle_SYS_ROOT_RECEIVED('root_uuid')
+        self.vm._got_root('root_uuid')
         self.vm.handle_SV_SHARE_CHANGED(info=share_holder)
         shouldbe_dir = os.path.join(self.shares_dir,
                                     "share from " + u"Ramón".encode("utf8"))
@@ -654,6 +689,7 @@
                                    free_bytes, suggested_path)
         udf = UDF.from_udf_volume(volume, path)
         udf.subscribed = subscribed
+        udf.generation = generation
         return udf, volume
 
     def test_build_udf_path(self):
@@ -715,20 +751,16 @@
         suggested_path = "suggested_path"
         udf, volume = self._create_udf(uuid.uuid4(), 'udf_node_id',
                                        '~/' + suggested_path, subscribed=True)
-        query_d = defer.Deferred()
-        space_d = defer.Deferred()
-        query_d.addCallback(lambda _: space_d)
-        def fake_query(data):
-            self.assertEquals([(udf.volume_id, udf.node_id, '')], data)
-            query_d.callback(None)
-        def fake_inquire_free_space(volume_id):
-            self.assertEquals(volume_id, request.ROOT)
-            space_d.callback(None)
-        self.main.action_q.query = fake_query
-        self.main.action_q.inquire_free_space = fake_inquire_free_space
+        get_delta_d = defer.Deferred()
+        def fake_get_delta(volume_id, generation):
+            """A fake get_delta that check the arguments."""
+            self.assertEquals(udf.volume_id, volume_id)
+            self.assertEquals(udf.generation, generation)
+            get_delta_d.callback(None)
+        self.main.action_q.get_delta = fake_get_delta
 
         yield self.vm.add_udf(udf)
-        yield query_d
+        yield get_delta_d
         self.assertEquals(os.path.join(self.home_dir, suggested_path),
                           udf.path)
         self.assertEquals(1, len(self.vm.udfs))
@@ -849,8 +881,9 @@
         self.assertEquals('udf_uuid', udf.node_id)
         self.assertEquals(os.path.join(self.home_dir, 'UDF'), udf.path)
         # check that the root it's there
-        self.assertTrue('' in self.vm.shares)
-        self.assertEquals(self.vm.shares[''].node_id, str(root_volume.node_id))
+        self.assertTrue(request.ROOT in self.vm.shares)
+        self.assertEquals(self.vm.shares[request.ROOT].node_id,
+                          str(root_volume.node_id))
         # now send the same list again and check
         # use a custom home
         with environ('HOME', self.home_dir):
@@ -896,7 +929,7 @@
         udf_id = uuid.uuid4()
         udf_volume = volumes.UDFVolume(udf_id, 'udf_uuid', None, 10, u'~/ñoño')
         # initialize the the root
-        self.vm.handle_SYS_ROOT_RECEIVED('root_uuid')
+        self.vm._got_root('root_uuid')
         response = [share_volume, udf_volume]
         d = defer.Deferred()
         self._listen_for('VM_UDF_CREATED', d.callback)
@@ -927,8 +960,9 @@
             self.vm.handle_AQ_LIST_VOLUMES(response)
         self.assertEquals(1, len(self.vm.shares)) # the new share and root
         # check that the root is in the shares dict
-        self.assertTrue('' in self.vm.shares)
-        self.assertEquals(self.vm.shares[''].node_id, str(root_volume.node_id))
+        self.assertTrue(request.ROOT in self.vm.shares)
+        self.assertEquals(self.vm.shares[request.ROOT].node_id,
+                          str(root_volume.node_id))
 
     def test_get_udf_path_name(self):
         """Test for _get_udf_path_name."""
@@ -1413,7 +1447,7 @@
                                            'username', 'visible_username',
                                            True, 'View')
         # initialize the the root
-        self.vm.handle_SYS_ROOT_RECEIVED('root_uuid')
+        self.vm._got_root('root_uuid')
         self._listen_for('VM_UDF_CREATED', d.callback)
         def check_udf(info):
             """The udf creation callback"""
@@ -1475,7 +1509,7 @@
         yield self.vm.add_udf(udf)
         self.vm.add_share(share)
         # initialize the the root
-        self.vm.handle_SYS_ROOT_RECEIVED('root_uuid')
+        self.vm._got_root('root_uuid')
         d = defer.Deferred()
         self._listen_for('VM_VOLUME_DELETED', d.callback)
         with environ('HOME', self.home_dir):
@@ -1585,7 +1619,7 @@
     def test_no_UDFs_inside_root(self):
         """Test that a UDF can't be created inside the root"""
         # initialize the root
-        self.vm.handle_SYS_ROOT_RECEIVED('root_uuid')
+        self.vm._got_root('root_uuid')
         udf_path = os.path.join(self.root_dir, 'udf_inside_root')
         # patch FakeAQ
         def create_udf(path, name, marker):
@@ -1695,8 +1729,8 @@
         share_path = os.path.join(self.main.shares_dir, dir_name)
         share = Share.from_share_volume(share_volume, share_path)
         # get the root
-        root = self.vm.get_volume('')
-        self.vm.handle_SYS_ROOT_RECEIVED('root_uuid')
+        root = self.vm.get_volume(request.ROOT)
+        self.vm._got_root('root_uuid')
         # create a UDF
         udf_id = uuid.uuid4()
         udf, volume = self._create_udf(udf_id, 'udf_node_id', '~/UDF')
@@ -1738,8 +1772,8 @@
         share_path = os.path.join(self.main.shares_dir, dir_name)
         share = Share.from_share_volume(share_volume, share_path)
         # get the root
-        root = self.vm.get_volume('')
-        self.vm.handle_SYS_ROOT_RECEIVED('root_node_id')
+        root = self.vm.get_volume(request.ROOT)
+        self.vm._got_root('root_node_id')
         # create a UDF
         udf_id = uuid.uuid4()
         udf, volume = self._create_udf(udf_id, 'udf_node_id', '~/UDF')
@@ -1763,7 +1797,7 @@
     def test_UDF_cant_be_a_symlink(self):
         """Test that a UDF can't be a symlink."""
         # initialize the root
-        self.vm.handle_SYS_ROOT_RECEIVED('root_uuid')
+        self.vm._got_root('root_uuid')
         real_udf_path = os.path.join(self.home_dir, "my_udf")
         udf_path = os.path.join(self.home_dir, "MyUDF")
         # patch FakeAQ
@@ -1789,7 +1823,7 @@
     def test_UDF_cant_be_inside_symlink(self):
         """Test that a UDF can't be inside a symlink."""
         # initialize the root
-        self.vm.handle_SYS_ROOT_RECEIVED('root_uuid')
+        self.vm._got_root('root_uuid')
         real_udf_path = os.path.join(self.home_dir, "udf_parent", "my_udf")
         udf_path = os.path.join(self.home_dir, "MyUDF")
         # patch FakeAQ
@@ -1839,10 +1873,10 @@
                            for event in events)
         self.assertIn(str(share_id), events_dict)
         self.assertIn(str(udf_id), events_dict)
-        self.assertIn('', events_dict)
+        self.assertIn(request.ROOT, events_dict)
         self.assertEquals(1, events_dict[str(share_id)])
         self.assertEquals(1, events_dict[str(udf_id)])
-        self.assertEquals(1, events_dict[''])
+        self.assertEquals(1, events_dict[request.ROOT])
 
     @defer.inlineCallbacks
     def test_server_rescan_error(self):
@@ -1888,10 +1922,10 @@
                            for event in events)
         self.assertIn(str(share_id), events_dict)
         self.assertIn(str(udf_id), events_dict)
-        self.assertIn('', events_dict)
+        self.assertIn(request.ROOT, events_dict)
         self.assertEquals(1, events_dict[str(share_id)])
         self.assertEquals(1, events_dict[str(udf_id)])
-        self.assertEquals(1, events_dict[''])
+        self.assertEquals(1, events_dict[request.ROOT])
         # set the local metadata generation to new value
         share = self.vm.shares[str(share_id)]
         share.generation = share_volume.generation
@@ -1901,7 +1935,7 @@
         self.vm.udfs[str(udf_id)] = udf
         root = self.vm.root
         root.generation = root_volume.generation
-        self.vm.shares[''] = root
+        self.vm.shares[request.ROOT] = root
         # now that we have the volumes in metadata, try with a higher  value.
         share_volume.generation = 10
         udf_volume.generation = 5
@@ -1916,7 +1950,7 @@
                            for event in events)
         self.assertIn(str(share_id), events_dict)
         self.assertIn(str(udf_id), events_dict)
-        self.assertNotIn('', events_dict) # same generartion as metadata
+        self.assertNotIn(request.ROOT, events_dict) # same generartion as metadata
         self.assertEquals(10, events_dict[str(share_id)])
         self.assertEquals(5, events_dict[str(udf_id)])
         # now only change the root volume generation
@@ -1933,8 +1967,8 @@
                            for event in events)
         self.assertNotIn(str(share_id), events_dict)
         self.assertNotIn(str(udf_id), events_dict)
-        self.assertIn('', events_dict) # same generartion as metadata
-        self.assertEquals(100, events_dict[''])
+        self.assertIn(request.ROOT, events_dict) # same generartion as metadata
+        self.assertEquals(100, events_dict[request.ROOT])
 
     @defer.inlineCallbacks
     def test_update_generation(self):
@@ -1948,8 +1982,8 @@
         share_path = os.path.join(self.main.shares_dir, dir_name)
         share = Share.from_share_volume(share_volume, share_path)
         # get the root
-        root = self.vm.get_volume('')
-        self.vm.handle_SYS_ROOT_RECEIVED('root_node_id')
+        root = self.vm.get_volume(request.ROOT)
+        self.vm._got_root('root_node_id')
         # create a UDF
         udf_id = uuid.uuid4()
         udf, _ = self._create_udf(udf_id, 'udf_node_id', '~/UDF')
@@ -1971,6 +2005,205 @@
         self.assertRaises(VolumeDoesNotExist, self.vm.update_generation,
                           str(uuid.uuid4()), 1)
 
+    @defer.inlineCallbacks
+    def test_handle_SV_VOLUME_NEW_GENERATION_udf(self):
+        """Test handle_SV_VOLUME_NEW_GENERATION for udf."""
+        # create a UDF
+        udf_id = uuid.uuid4()
+        udf, udf_volume = self._create_udf(udf_id, 'udf_node_id', '~/UDF')
+        yield self.vm.add_udf(udf)
+        self.vm.update_generation(udf.volume_id, 10)
+        d = defer.Deferred()
+        self.patch(self.main.action_q, 'get_delta', lambda v, g: d.callback((v, g)))
+        self.main.event_q.push('SV_VOLUME_NEW_GENERATION', udf.volume_id, 100)
+        vol_id, gen = yield d
+        vol = self.vm.get_volume(vol_id)
+        self.assertEqual(vol_id, vol.volume_id)
+        self.assertEqual(gen, vol.generation)
+
+    @defer.inlineCallbacks
+    def test_handle_SV_VOLUME_NEW_GENERATION_udf_from_scratch(self):
+        """Test handle_SV_VOLUME_NEW_GENERATION for udf."""
+        # create a UDF
+        udf_id = uuid.uuid4()
+        udf, udf_volume = self._create_udf(udf_id, 'udf_node_id', '~/UDF')
+        yield self.vm.add_udf(udf)
+        self.vm.update_generation(udf.volume_id, None)
+        d = defer.Deferred()
+        self.patch(self.main.action_q, 'rescan_from_scratch', d.callback)
+        self.main.event_q.push('SV_VOLUME_NEW_GENERATION', udf.volume_id, 100)
+        vol_id = yield d
+        self.assertEquals(vol_id, udf.volume_id)
+
+    @defer.inlineCallbacks
+    def test_handle_SV_VOLUME_NEW_GENERATION_udf_eq(self):
+        """Test handle_SV_VOLUME_NEW_GENERATION for udf."""
+        # get the root
+        udf_id = uuid.uuid4()
+        udf, udf_volume = self._create_udf(udf_id, 'udf_node_id', '~/UDF')
+        yield self.vm.add_udf(udf)
+        self.vm.update_generation(udf.volume_id, 100)
+        handler = MementoHandler()
+        handler.setLevel(logging.INFO)
+        self.vm.log.addHandler(handler)
+        self.main.event_q.push('SV_VOLUME_NEW_GENERATION', udf.volume_id, 100)
+        self.assertEqual(1, len(handler.records))
+        msg = 'Got SV_VOLUME_NEW_GENERATION(%r, %r) but volume' + \
+                ' is at generation: %r'
+        self.assertEqual(msg % (udf.volume_id, 100, 100),
+                         handler.records[0].message)
+        self.vm.log.removeHandler(handler)
+
+    @defer.inlineCallbacks
+    def test_handle_SV_VOLUME_NEW_GENERATION_root(self):
+        """Test handle_SV_VOLUME_NEW_GENERATION for root share."""
+        # get the root
+        root = self.vm.get_volume(request.ROOT)
+        self.vm._got_root('root_node_id')
+        self.vm.update_generation(root.volume_id, 10)
+        d = defer.Deferred()
+        self.patch(self.main.action_q, 'get_delta', lambda v, g: d.callback((v, g)))
+        self.main.event_q.push('SV_VOLUME_NEW_GENERATION', root.volume_id, 100)
+        vol_id, gen = yield d
+        vol = self.vm.get_volume(vol_id)
+        self.assertEqual(vol_id, vol.volume_id)
+        self.assertEqual(gen, vol.generation)
+
+    @defer.inlineCallbacks
+    def test_handle_SV_VOLUME_NEW_GENERATION_root_from_scratch(self):
+        """Test handle_SV_VOLUME_NEW_GENERATION for root share."""
+        # get the root
+        root = self.vm.get_volume(request.ROOT)
+        self.vm._got_root('root_node_id')
+        self.vm.update_generation(root.volume_id, None)
+        d = defer.Deferred()
+        self.patch(self.main.action_q, 'rescan_from_scratch', d.callback)
+        self.main.event_q.push('SV_VOLUME_NEW_GENERATION', root.volume_id, 100)
+        vol_id = yield d
+        self.assertEquals(vol_id, root.volume_id)
+
+    def test_handle_SV_VOLUME_NEW_GENERATION_root_eq(self):
+        """Test handle_SV_VOLUME_NEW_GENERATION for root share."""
+        # get the root
+        root = self.vm.get_volume(request.ROOT)
+        self.vm._got_root('root_node_id')
+        self.vm.update_generation(root.volume_id, 100)
+        handler = MementoHandler()
+        handler.setLevel(logging.INFO)
+        self.vm.log.addHandler(handler)
+        self.main.event_q.push('SV_VOLUME_NEW_GENERATION', root.volume_id, 100)
+        self.assertEqual(1, len(handler.records))
+        msg = 'Got SV_VOLUME_NEW_GENERATION(%r, %r) but volume' + \
+                ' is at generation: %r'
+        self.assertEqual(msg % (root.volume_id, 100, 100),
+                         handler.records[0].message)
+        self.vm.log.removeHandler(handler)
+
+    @defer.inlineCallbacks
+    def test_handle_SV_VOLUME_NEW_GENERATION_share(self):
+        """Test handle_SV_VOLUME_NEW_GENERATION for a share."""
+        share_id = uuid.uuid4()
+        share_volume = volumes.ShareVolume(share_id, 'fake_share_uuid', None,
+                                           10, 'to_me', 'fake_share', 'username',
+                                           'visible_username', True, 'View')
+        dir_name = self.vm._build_share_path(share_volume.share_name,
+                                          share_volume.other_visible_name)
+        share_path = os.path.join(self.main.shares_dir, dir_name)
+        share = Share.from_share_volume(share_volume, share_path)
+        self.vm.add_share(share)
+        self.vm.update_generation(share.volume_id, 10)
+        d = defer.Deferred()
+        self.patch(self.main.action_q, 'get_delta', lambda v, g: d.callback((v, g)))
+        self.main.event_q.push('SV_VOLUME_NEW_GENERATION', share.volume_id, 100)
+        vol_id, gen = yield d
+        vol = self.vm.get_volume(vol_id)
+        self.assertEqual(vol_id, vol.volume_id)
+        self.assertEqual(gen, vol.generation)
+
+    @defer.inlineCallbacks
+    def test_handle_SV_VOLUME_NEW_GENERATION_share_from_scratch(self):
+        """Test handle_SV_VOLUME_NEW_GENERATION for a share."""
+        share_id = uuid.uuid4()
+        share_volume = volumes.ShareVolume(share_id, 'fake_share_uuid', None,
+                                           10, 'to_me', 'fake_share', 'username',
+                                           'visible_username', True, 'View')
+        dir_name = self.vm._build_share_path(share_volume.share_name,
+                                          share_volume.other_visible_name)
+        share_path = os.path.join(self.main.shares_dir, dir_name)
+        share = Share.from_share_volume(share_volume, share_path)
+        self.vm.add_share(share)
+        self.vm.update_generation(share.volume_id, None)
+        d = defer.Deferred()
+        self.patch(self.main.action_q, 'rescan_from_scratch', d.callback)
+        self.main.event_q.push('SV_VOLUME_NEW_GENERATION', share.volume_id, 100)
+        vol_id = yield d
+        self.assertEquals(vol_id, share.volume_id)
+
+    def test_handle_SV_VOLUME_NEW_GENERATION_share_eq(self):
+        """Test handle_SV_VOLUME_NEW_GENERATION for a share."""
+        share_id = uuid.uuid4()
+        share_volume = volumes.ShareVolume(share_id, 'fake_share_uuid', None,
+                                           10, 'to_me', 'fake_share', 'username',
+                                           'visible_username', True, 'View')
+        dir_name = self.vm._build_share_path(share_volume.share_name,
+                                          share_volume.other_visible_name)
+        share_path = os.path.join(self.main.shares_dir, dir_name)
+        share = Share.from_share_volume(share_volume, share_path)
+        self.vm.add_share(share)
+        self.vm.update_generation(share.volume_id, 100)
+        handler = MementoHandler()
+        handler.setLevel(logging.INFO)
+        self.vm.log.addHandler(handler)
+        self.main.event_q.push('SV_VOLUME_NEW_GENERATION', share.volume_id, 100)
+        self.assertEqual(1, len(handler.records))
+        msg = 'Got SV_VOLUME_NEW_GENERATION(%r, %r) but volume' + \
+                ' is at generation: %r'
+        self.assertEqual(msg % (share.volume_id, 100, 100),
+                         handler.records[0].message)
+        self.vm.log.removeHandler(handler)
+
+    @defer.inlineCallbacks
+    def test_handle_AQ_DELTA_NOT_POSSIBLE(self):
+        """Test for handle_AQ_DELTA_NOT_POSSIBLE."""
+        share_id = uuid.uuid4()
+        share_volume = volumes.ShareVolume(share_id, 'fake_share_uuid', None,
+                                           10, 'to_me', 'fake_share', 'username',
+                                           'visible_username', True, 'View')
+        dir_name = self.vm._build_share_path(share_volume.share_name,
+                                          share_volume.other_visible_name)
+        share_path = os.path.join(self.main.shares_dir, dir_name)
+        share = Share.from_share_volume(share_volume, share_path)
+        # get the root
+        root = self.vm.get_volume(request.ROOT)
+        self.vm._got_root('root_node_id')
+        # create a UDF
+        udf_id = uuid.uuid4()
+        udf, _ = self._create_udf(udf_id, 'udf_node_id', '~/UDF')
+        self.vm.add_share(share)
+        yield self.vm.add_udf(udf)
+        # patch AQ.rescan_from_scratch
+        calls = []
+        self.patch(self.main.action_q, 'rescan_from_scratch', calls.append)
+        self.vm.handle_AQ_DELTA_NOT_POSSIBLE(udf_id)
+        self.vm.handle_AQ_DELTA_NOT_POSSIBLE(share_id)
+        self.vm.handle_AQ_DELTA_NOT_POSSIBLE(root.volume_id)
+        for i, vol in enumerate([udf, share, root]):
+            self.assertEquals(calls[i], str(vol.volume_id))
+
+    def test_handle_AQ_DELTA_NOT_POSSIBLE_missing_volume(self):
+        """Test for handle_AQ_DELTA_NOT_POSSIBLE with an missing volume."""
+        handler = MementoHandler()
+        handler.setLevel(logging.WARNING)
+        self.vm.log.addHandler(handler)
+        vol_id = uuid.uuid4()
+        try:
+            self.vm.handle_AQ_DELTA_NOT_POSSIBLE(vol_id)
+        finally:
+            self.vm.log.removeHandler(handler)
+        self.assertEqual(1, len(handler.records))
+        msg = 'Got a AQ_DELTA_NOT_POSSIBLE for a missing volume: %r'
+        self.assertEqual(msg % str(vol_id), handler.records[0].message)
+
 
 class MetadataTestCase(BaseTwistedTestCase):
     md_version_None = False
@@ -2062,7 +2295,7 @@
         # add the root_uuid key
         root_share = _Share(path=self.root_dir)
         root_share.access_level = 'Modify'
-        old_shelf[''] = root_share
+        old_shelf[request.ROOT] = root_share
         for idx in range(1, 10):
             sid = str(uuid.uuid4())
             old_shelf[sid] =  _Share(path=os.path.join(self.shares_dir, str(idx)),
@@ -2129,7 +2362,7 @@
                              self.data_dir, self.partials_dir)
         new_keys = [new_key for new_key in self.main.vm.shares.keys()]
         self.assertEquals(2, len(new_keys)) # the fake share plus root
-        for key in ['', share.id]:
+        for key in [request.ROOT, share.id]:
             self.assertIn(key, new_keys)
         self.check_version()
 
@@ -2322,7 +2555,7 @@
         # add the root_uuid key
         root_share = _Share(path=self.root_dir)
         root_share.access_level = 'Modify'
-        maybe_old_shelf[''] = root_share
+        maybe_old_shelf[request.ROOT] = root_share
         for idx in range(1, 10):
             share_id = str(uuid.uuid4())
             maybe_old_shelf[share_id] = \
@@ -2483,7 +2716,7 @@
         root_share.access_level = 'Modify'
         # set None to the share path
         root_share.path = None
-        shares[''] = root_share
+        shares[request.ROOT] = root_share
 
         if self.md_version_None:
             self.set_md_version('')
@@ -2521,9 +2754,9 @@
         self.set_md_version('5')
         # create some old shares and shared metadata
         legacy_shares = LegacyShareFileShelf(self.share_md_dir)
-        root_share = _Share(path=self.root_dir, share_id='',
+        root_share = _Share(path=self.root_dir, share_id=request.ROOT,
                             access_level='Modify')
-        legacy_shares[''] = root_share
+        legacy_shares[request.ROOT] = root_share
         for idx, name in enumerate(['share'] * 10):
             sid = str(uuid.uuid4())
             share_name = name + '_' + str(idx)
@@ -2595,9 +2828,9 @@
         self.udfs_md_dir = os.path.join(self.vm_data_dir, 'udfs')
         # create some old shares and shared metadata
         legacy_shares = LegacyShareFileShelf(self.share_md_dir)
-        root_share = _Share(path=self.root_dir, share_id='',
+        root_share = _Share(path=self.root_dir, share_id=request.ROOT,
                             access_level='Modify')
-        legacy_shares[''] = root_share
+        legacy_shares[request.ROOT] = root_share
         for idx, name in enumerate(['share'] * 10):
             sid = str(uuid.uuid4())
             share_name = name + '_' + str(idx)
@@ -2695,9 +2928,9 @@
         self.udfs_md_dir = os.path.join(self.vm_data_dir, 'udfs')
         # create some old shares and shared metadata
         legacy_shares = LegacyShareFileShelf(self.share_md_dir)
-        root_share = _Share(path=self.root_dir, share_id='',
+        root_share = _Share(path=self.root_dir, share_id=request.ROOT,
                             access_level='Modify', node_id=str(uuid.uuid4()))
-        legacy_shares[''] = root_share
+        legacy_shares[request.ROOT] = root_share
         for idx, name in enumerate(['share'] * 3):
             sid = str(uuid.uuid4())
             share_name = name + '_' + str(idx)
@@ -2802,9 +3035,9 @@
         self.set_md_version('5')
         # create some old shares and shared metadata
         legacy_shares = LegacyShareFileShelf(self.share_md_dir)
-        root_share = _Share(path=self.root_dir, share_id='',
+        root_share = _Share(path=self.root_dir, share_id=request.ROOT,
                             access_level='Modify')
-        legacy_shares[''] = root_share
+        legacy_shares[request.ROOT] = root_share
         for idx, name in enumerate(['share'] * 10):
             sid = str(uuid.uuid4())
             share_name = name + '_' + str(idx)

=== modified file 'ubuntuone/syncdaemon/action_queue.py'
--- ubuntuone/syncdaemon/action_queue.py	2010-07-19 20:15:06 +0000
+++ ubuntuone/syncdaemon/action_queue.py	2010-07-22 13:39:48 +0000
@@ -703,14 +703,6 @@
         self.connector = None
         self.connect_in_progress = False
 
-    def _node_state_callback(self, share_id, node_id, hash):
-        """Called by the client when notified that node changed."""
-        volume = self.main.vm.udfs.get(share_id, None)
-        if volume is not None and not volume.subscribed:
-            return
-        self.event_queue.push('SV_HASH_NEW',
-                              share_id=share_id, node_id=node_id, hash=hash)
-
     def _share_change_callback(self, info):
         """Called by the client when notified that a share changed."""
         self.event_queue.push('SV_SHARE_CHANGED', info=info)
@@ -827,7 +819,6 @@
         # does nothing (safely).
         self.client = ThrottlingStorageClientFactory.buildProtocol(self, addr)
 
-        self.client.set_node_state_callback(self._node_state_callback)
         self.client.set_share_change_callback(self._share_change_callback)
         self.client.set_share_answer_callback(self._share_answer_callback)
         self.client.set_free_space_callback(self._free_space_callback)
@@ -878,6 +869,7 @@
     def _send_request_and_handle_errors(self, request, request_error,
                                         event_error, event_ok,
                                         fire_deferred=True,
+                                        handle_exception=True,
                                         args=(), kwargs={}):
         """Send 'request' to the server, using params 'args' and 'kwargs'.
 
@@ -919,8 +911,11 @@
             event = 'SYS_SERVER_ERROR'
             self.event_queue.push(event, error=str(failure))
         except Exception, failure:
-            event = 'SYS_UNKNOWN_ERROR'
-            self.event_queue.push(event)
+            if handle_exception:
+                event = 'SYS_UNKNOWN_ERROR'
+                self.event_queue.push(event)
+            else:
+                raise
         else:
             logger.info("The request '%s' finished OK.", req_name)
             if event_ok is not None:
@@ -997,50 +992,19 @@
             # callback the deferred if everything went ok
             self.deferred.callback(self.client)
 
-    def server_rescan(self, root_mdid, data_gen):
-        """Do the server rescan."""
-
-        @defer.inlineCallbacks
-        def _get_root_and_query(root_mdid, data_gen):
-            """Get user's root and then query each element in data_gen()."""
-            yield self.get_root(root_mdid)
-            data = data_gen()
-            logger.info("Server rescan: will query %d objects", len(data))
-            # check we're going to actually log, because this could be expensive
-            if logger.isEnabledFor(TRACE):
-                for share, node, hash in data:
-                    logger.trace("Server rescan: share: %r, node: %r, hash: %s",
-                                 share or '/root/', node, hash)
-                logger.trace("Server rescan: all data shown")
-            yield self.client.query(data)
-
-        get_root_and_query_d = self._send_request_and_handle_errors(
-            request=_get_root_and_query,
+    @defer.inlineCallbacks
+    def query_volumes(self):
+        """Get the list of volumes.
+
+        This method will *not* queue a command, the request will be
+        executed right away.
+        """
+        result = yield self._send_request_and_handle_errors(
+            request=self.client.list_volumes,
             request_error=None, event_error=None,
-            event_ok='SYS_SERVER_RESCAN_DONE', fire_deferred=False,
-            args=(root_mdid, data_gen)
-        )
-        return get_root_and_query_d
-
-    def get_root(self, marker):
-        """Get the user's root uuid.
-
-        Use the uuid_map, so the caller can use the marker in followup
-        operations.
-
-        """
-        log = mklog(logger, 'get_root', '', marker, marker=marker)
-        log.debug('starting')
-        d = self.client.get_root()
-        d.addCallbacks(*log.callbacks())
-        d.addCallbacks(passit(lambda root: self.uuid_map.set(marker, root)),
-                       passit(lambda f: self.uuid_map.err(marker, f)))
-        def handle_root(root_id):
-            """Push SYS_ROOT_RECEIVED event"""
-            self.event_queue.push('SYS_ROOT_RECEIVED', root_id=root_id)
-            return root_id
-        d.addCallback(handle_root)
-        return d
+            event_ok=None, fire_deferred=False,
+            handle_exception=False)
+        defer.returnValue(result.volumes)
 
     def make_file(self, share_id, parent_id, name, marker):
         """
@@ -1134,17 +1098,6 @@
         """
         return ListVolumes(self.meta_queue).start()
 
-    def query_volumes(self):
-        """List all the volumes the user has.
-
-        This includes the volumes:
-            - all the user's UDFs.
-            - all the shares the user has accepted.
-            - the root-root volume.
-
-        """
-        return VolumesQuery(self.meta_queue).start()
-
     def delete_volume(self, volume_id):
         """Delete a volume on the server, removing the associated tree.
 
@@ -1256,6 +1209,18 @@
         """
         return GetDelta(self.meta_queue, volume_id, generation).start()
 
+    def rescan_from_scratch(self, volume_id):
+        """Get a delta from scratch for the volume.
+
+        @param volume_id: the id of the volume to get the delta.
+
+        Result will be signaled using events:
+            - AQ_RESCAN_FROM_SCRATCH_OK on succeess.
+            - AQ_RESCAN_FROM_SCRATCH_ERROR on generic failure.
+        """
+        # TODO: implement rescan from scratch
+        return None
+
 
 SKIP_THIS_ITEM = object()
 
@@ -1989,18 +1954,6 @@
                                            error=failure.getErrorMessage())
 
 
-class VolumesQuery(ListVolumes):
-    """List all the volumes for a given user, but doesn't push a event."""
-
-    def handle_success(self, success):
-        """It worked!, just forward the value"""
-        return success.volumes
-
-    def handle_failure(self, failure):
-        """It didn't work! just forward the failure."""
-        return failure
-
-
 class DeleteVolume(ActionQueueCommand):
     """Delete an exsistent volume."""
 
@@ -2065,6 +2018,7 @@
     def handle_success(self, request):
         """It worked! Push the success event."""
         data = dict(
+            volume_id=self.volume_id,
             delta_content=request.response,
             end_generation=request.end_generation,
             full=request.full,

=== modified file 'ubuntuone/syncdaemon/event_queue.py'
--- ubuntuone/syncdaemon/event_queue.py	2010-07-19 20:15:06 +0000
+++ ubuntuone/syncdaemon/event_queue.py	2010-07-22 13:39:48 +0000
@@ -85,19 +85,14 @@
     'AQ_CHANGE_PUBLIC_ACCESS_ERROR': ('share_id', 'node_id', 'error'),
     'AQ_PUBLIC_FILES_LIST_OK': ('public_files',),
     'AQ_PUBLIC_FILES_LIST_ERROR': ('error',),
-    'AQ_DELTA_OK': ('delta_content', 'end_generation', 'full', 'free_bytes'),
+    'AQ_DELTA_OK': ('volume_id', 'delta_content', 'end_generation',
+                    'full', 'free_bytes'),
     'AQ_DELTA_ERROR': ('volume_id', 'error'),
     'AQ_DELTA_NOT_POSSIBLE': ('volume_id',),
 
     'SV_SHARE_CHANGED': ('info',),
     'SV_SHARE_DELETED': ('share_id',),
     'SV_SHARE_ANSWERED': ('share_id', 'answer'),
-    'SV_HASH_NEW': ('share_id', 'node_id', 'hash'),
-    'SV_FILE_NEW': ('share_id', 'node_id', 'parent_id', 'name'),
-    'SV_DIR_NEW': ('share_id', 'node_id', 'parent_id', 'name'),
-    'SV_FILE_DELETED': ('share_id', 'node_id'),
-    'SV_MOVED': ('share_id', 'node_id', 'new_share_id', 'new_parent_id',
-                 'new_name'),
     'SV_FREE_SPACE': ('share_id', 'free_bytes'),
     'SV_ACCOUNT_CHANGED': ('account_info',),
     'SV_VOLUME_CREATED': ('volume',),

=== modified file 'ubuntuone/syncdaemon/main.py'
--- ubuntuone/syncdaemon/main.py	2010-03-19 14:28:40 +0000
+++ ubuntuone/syncdaemon/main.py	2010-07-22 13:39:48 +0000
@@ -233,9 +233,7 @@
 
     def server_rescan(self):
         """Do the server rescan."""
-        mdobj = self.fs.get_by_path(self.root_dir)
-        self.action_q.server_rescan(mdobj.mdid,
-                                    self.fs.get_data_for_server_rescan)
+        return self.vm.server_rescan()
 
     def set_oauth_token(self, key, secret):
         """ Sets the oauth token """

=== modified file 'ubuntuone/syncdaemon/sync.py'
--- ubuntuone/syncdaemon/sync.py	2010-07-20 16:17:58 +0000
+++ ubuntuone/syncdaemon/sync.py	2010-07-22 13:39:48 +0000
@@ -24,11 +24,9 @@
 import sys
 
 from ubuntuone.syncdaemon.marker import MDMarker
-from ubuntuone.storageprotocol.dircontent_pb2 import DIRECTORY
-from ubuntuone.storageprotocol import dircontent
+from ubuntuone.storageprotocol import delta
 from ubuntuone.syncdaemon.fsm.fsm import \
     StateMachineRunner, StateMachine
-from ubuntuone.syncdaemon.interfaces import IMarker
 from ubuntuone.syncdaemon import u1fsfsm
 from ubuntuone.syncdaemon.logger import DebugCapture
 from ubuntuone.syncdaemon.filesystem_manager import InconsistencyError
@@ -309,6 +307,7 @@
         with DebugCapture(self.log.logger):
             func_name = super(SyncStateMachineRunner, self).on_event(*args,
                                                                      **kwargs)
+
             if not is_debug:
                 self.log.info("Called %s (In: %s)" % (func_name, in_state))
 
@@ -362,8 +361,18 @@
             is_directory=self.key.is_directory(),
         )
 
-    def check_new_volume_generation(self, volume_id, new_generation):
-        """Check the new generation for the volume and take proper action."""
+    def update_generation(self, volume_id, node_id, new_generation):
+        """Update the generation for the node and volume."""
+
+        # update the file
+        try:
+            node = self.m.fs.get_by_node_id(volume_id, node_id)
+        except KeyError:
+            pass
+        else:
+            self.m.fs.set_by_mdid(node.mdid, generation=new_generation)
+
+        # update the volume
         try:
             volume = self.m.vm.get_volume(volume_id)
         except VolumeDoesNotExist:
@@ -405,7 +414,6 @@
         path = os.path.join(self.m.fs.get_abspath(share_id, mdobj.path), name)
         self.m.fs.create(path=path, share_id=share_id, node_id=node_id,
                          is_dir=True)
-        self.m.action_q.query([(share_id, node_id, "")])
         # pylint: disable-msg=W0704
         # this should be provided by FSM, fix!!
         try:
@@ -439,30 +447,6 @@
         self.m.fs.set_node_id(self.key['path'], node_id)
         self.reget_dir(event, params)
 
-    def reget_dir(self, event, params, hash=None):
-        """Reget the directory."""
-        self.m.action_q.cancel_download(share_id=self.key['share_id'],
-                            node_id=self.key['node_id'])
-        self.key.remove_partial()
-        self.m.action_q.query([(self.key['share_id'],
-                                self.key['node_id'],
-                                self.key['local_hash'] or "")])
-        self.key.set(server_hash=self.key['local_hash'])
-        self.key.sync()
-
-    def get_dir(self, event, params, hash):
-        """Get the directory."""
-        self.key.set(server_hash=hash)
-        self.key.sync()
-        self.m.fs.create_partial(node_id=self.key['node_id'],
-                                 share_id=self.key['share_id'])
-        self.m.action_q.listdir(
-            self.key['share_id'], self.key['node_id'], hash,
-            lambda : self.m.fs.get_partial_for_writing(
-                node_id=self.key['node_id'],
-                share_id=self.key['share_id'])
-            )
-
     def file_conflict(self, event, params, hash, crc32, size, stat):
         """This file is in conflict."""
         self.key.move_to_conflict()
@@ -474,92 +458,6 @@
                             node_id=self.key['node_id'])
         self.get_file(event, params, hash)
 
-    def merge_directory(self, event, params, hash):
-        """Merge the server directory with the local one."""
-        new_files = []
-        new_dirs = []
-        deleted_filedirs = []
-        moved = set()
-
-        try:
-            fd = self.m.fs.get_partial(node_id=self.key['node_id'],
-                            share_id=self.key['share_id'])
-        except InconsistencyError:
-            self.key.remove_partial()
-            self.key.set(server_hash=self.key['local_hash'])
-            self.key.sync()
-            self.m.action_q.query([
-                (self.key["share_id"], self.key["node_id"], "")])
-            # we dont perform the merge, we try to re get it
-            return
-
-
-        items = dircontent.parse_dir_content(fd)
-        server_dir = [ (o.utf8_name, o.node_type == DIRECTORY, o.uuid)
-                        for o in items ]
-        client_dir = self.m.fs.dir_content(self.key['path'])
-        # XXX: lucio.torre: with huge dirs, this could take a while
-
-        share = self.key['share_id']
-        for name, isdir, uuid in server_dir:
-            # we took the name as bytes already encoded in utf8
-            # directly from dircontent!
-            try:
-                md = self.m.fs.get_by_node_id(share, uuid)
-            except KeyError:
-                # not there, a new thing
-                if isdir:
-                    new_dirs.append((share, uuid, name))
-                else:
-                    new_files.append((share, uuid, name))
-                continue
-            mdpath = self.m.fs.get_abspath(md.share_id, md.path)
-            if mdpath != os.path.join(self.key['path'], name):
-                # this was moved, or maybe the server still didn't receive
-                # the move that happened here
-                if not self.m.action_q.node_is_with_queued_move(share, uuid):
-                    # mark as moved
-                    moved.add(uuid)
-                    # signal moved
-                    self.m.event_q.push("SV_MOVED",
-                        share_id=md.share_id, node_id=uuid,
-                        new_share_id=share, new_parent_id=self.key['node_id'],
-                        new_name=name)
-
-
-        for name, isdir, uuid in client_dir:
-            if uuid is None:
-                continue
-
-            if not (name, isdir, uuid) in server_dir:
-                # not there, a its gone on the server
-                if uuid in moved:
-                    # this was a move, dont delete
-                    continue
-                deleted_filedirs.append((share, uuid))
-
-        # these nodes are in process of being deleted from the server, so they
-        # are not exactly new locally (were already deleted here)
-        trash = set((share_id, node_id) for (share_id, node_id, _)
-                                        in self.m.fs.get_iter_trash())
-
-        parent_uuid = self.key['node_id']
-        for share, uuid in deleted_filedirs:
-            self.m.event_q.push("SV_FILE_DELETED",
-                                node_id=uuid, share_id=share)
-        for share, uuid, name in new_files:
-            if (share, uuid) not in trash:
-                self.m.event_q.push("SV_FILE_NEW", parent_id=parent_uuid,
-                                    node_id=uuid, share_id=share, name=name)
-        for share, uuid, name in new_dirs:
-            if (share, uuid) not in trash:
-                self.m.event_q.push("SV_DIR_NEW", parent_id=parent_uuid,
-                                    node_id=uuid, share_id=share, name=name)
-
-        self.key.remove_partial()
-        self.key.set(local_hash=hash)
-        self.key.sync()
-
     def new_file(self, event, params, share_id, node_id, parent_id, name):
         """create a local file."""
         mdobj = self.m.fs.get_by_node_id(share_id, parent_id)
@@ -569,7 +467,6 @@
         self.key.set(server_hash="")
         self.key.set(local_hash="")
         self.key.sync()
-        self.m.action_q.query([(share_id, node_id, "")])
 
     def new_file_on_server_with_local(self, event, params, share_id,
                                       node_id, parent_id, name):
@@ -616,15 +513,6 @@
             new_parent_id=new_parent_id, new_name=new_name)
         self.key.moved(self.key['share_id'], path_to)
 
-        # this is cheating, we change the state of another node
-        if not IMarker.providedBy(old_parent_id):
-            share_id = self.key['share_id']
-            self.m.action_q.cancel_download(share_id, old_parent_id)
-            old_parent.remove_partial()
-            self.m.fs.set_by_node_id(old_parent_id, share_id,
-                                     server_hash="", local_hash="")
-            self.m.action_q.query([(share_id, old_parent_id, "")])
-
         # we only hash it if we're a file, not a directory
         if not self.key.is_dir():
             self.m.hash_q.insert(self.key['path'], self.key['mdid'])
@@ -649,8 +537,7 @@
             self.key.remove_partial()
             self.key.set(server_hash=self.key['local_hash'])
             self.key.sync()
-            self.m.action_q.query([
-                (self.key["share_id"], self.key["node_id"], "")])
+
 
     def new_local_file(self, event, parms, path):
         """a new local file was created"""
@@ -690,10 +577,6 @@
         """Server answered that dir creation was ok."""
         self.m.fs.set_node_id(self.key['path'], new_id)
 
-        # query to get any updates in case dir was already there in server
-        self.m.action_q.query([(self.key['share_id'], self.key['node_id'],
-                                self.key['local_hash'])])
-
     def calculate_hash(self, event, params):
         """calculate the hash of this."""
         self.m.hash_q.insert(self.key['path'], self.key['mdid'])
@@ -809,12 +692,8 @@
     def dir_not_created_in_server(self, event, params, error):
         """Re-get the dir if it was because already exist."""
         if error == "ALREADY_EXISTS":
-            # delete metadata and query the father to converge
-            parent = self.m.fs.get_by_node_id(self.key['share_id'],
-                                              self.key['parent_id'])
+            # delete metadata
             self.key.delete_metadata()
-            self.m.action_q.query([(parent.share_id,
-                                    parent.node_id, parent.local_hash)])
         else:
             self.key.move_to_conflict()
             self.key.delete_metadata()
@@ -890,9 +769,6 @@
         self.key.remove_partial()
         self.key.set(server_hash=self.key['local_hash'])
         self.key.sync()
-        self.m.action_q.query([(self.key['share_id'],
-                                self.key['node_id'],
-                                self.key['local_hash'] or "")])
 
     # pylint: disable-msg=C0103
     def DESPAIR(self, event, params, *args, **kwargs):
@@ -927,14 +803,14 @@
         self.m = main
         self.m.event_q.subscribe(self)
 
-    def handle_SV_HASH_NEW(self, share_id, node_id, hash):
-        """on SV_HASH_NEW"""
+    def _handle_SV_HASH_NEW(self, share_id, node_id, hash):
+        """on SV_HASH_NEW.  No longer called by EQ, only internally."""
         key = FSKey(self.m.fs, share_id=share_id, node_id=node_id)
         log = FileLogger(self.logger, key)
         ssmr = SyncStateMachineRunner(self.fsm, self.m, key, log)
         ssmr.signal_event_with_hash("SV_HASH_NEW", hash)
 
-    def handle_SV_FILE_NEW(self, share_id, node_id, parent_id, name):
+    def _handle_SV_FILE_NEW(self, share_id, node_id, parent_id, name):
         """on SV_FILE_NEW"""
         parent = FSKey(self.m.fs, share_id=share_id, node_id=parent_id)
         path = os.path.join(parent["path"], name)
@@ -943,7 +819,7 @@
         ssmr = SyncStateMachineRunner(self.fsm, self.m, key, log)
         ssmr.on_event("SV_FILE_NEW", {}, share_id, node_id, parent_id, name)
 
-    def handle_SV_DIR_NEW(self, share_id, node_id, parent_id, name):
+    def _handle_SV_DIR_NEW(self, share_id, node_id, parent_id, name):
         """on SV_DIR_NEW"""
         parent = FSKey(self.m.fs, share_id=share_id, node_id=parent_id)
         path = os.path.join(parent["path"], name)
@@ -952,8 +828,8 @@
         ssmr = SyncStateMachineRunner(self.fsm, self.m, key, log)
         ssmr.on_event("SV_DIR_NEW", {}, share_id, node_id, parent_id, name)
 
-    def handle_SV_FILE_DELETED(self, share_id, node_id):
-        """on SV_FILE_DELETED"""
+    def _handle_SV_FILE_DELETED(self, share_id, node_id):
+        """on SV_FILE_DELETED. Not called by EQ anymore."""
         key = FSKey(self.m.fs, share_id=share_id, node_id=node_id)
         log = FileLogger(self.logger, key)
         ssmr = SyncStateMachineRunner(self.fsm, self.m, key, log)
@@ -1035,8 +911,8 @@
         key = FSKey(self.m.fs, mdid=marker)
         log = FileLogger(self.logger, key)
         ssmr = SyncStateMachineRunner(self.fsm, self.m, key, log)
-        ssmr.check_new_volume_generation(volume_id, new_generation)
         ssmr.on_event("AQ_FILE_NEW_OK", {}, new_id)
+        ssmr.update_generation(volume_id, new_id, new_generation)
 
     def handle_AQ_FILE_NEW_ERROR(self, marker, error):
         """on AQ_FILE_NEW_ERROR"""
@@ -1057,8 +933,8 @@
         key = FSKey(self.m.fs, mdid=marker)
         log = FileLogger(self.logger, key)
         ssmr = SyncStateMachineRunner(self.fsm, self.m, key, log)
-        ssmr.check_new_volume_generation(volume_id, new_generation)
         ssmr.on_event("AQ_DIR_NEW_OK", {}, new_id)
+        ssmr.update_generation(volume_id, new_id, new_generation)
 
     def handle_FS_FILE_CLOSE_WRITE(self, path):
         """on FS_FILE_CLOSE_WRITE"""
@@ -1082,7 +958,7 @@
         ssmr.on_event('HQ_HASH_ERROR', {})
 
     def handle_HQ_HASH_NEW(self, path, hash, crc32, size, stat):
-        """on HQ_HASH_NEW"""
+        """on HQ_HASH_NEW."""
         key = FSKey(self.m.fs, path=path)
         log = FileLogger(self.logger, key)
         ssmr = SyncStateMachineRunner(self.fsm, self.m, key, log)
@@ -1104,8 +980,8 @@
         key = FSKey(self.m.fs, share_id=share_id, node_id=node_id)
         log = FileLogger(self.logger, key)
         ssmr = SyncStateMachineRunner(self.fsm, self.m, key, log)
-        ssmr.check_new_volume_generation(share_id, new_generation)
         ssmr.signal_event_with_hash("AQ_UPLOAD_FINISHED", hash)
+        ssmr.update_generation(share_id, node_id, new_generation)
 
     def handle_AQ_UPLOAD_ERROR(self, share_id, node_id, error, hash):
         """on AQ_UPLOAD_ERROR"""
@@ -1114,7 +990,7 @@
         ssmr = SyncStateMachineRunner(self.fsm, self.m, key, log)
         ssmr.signal_event_with_error_and_hash("AQ_UPLOAD_ERROR", error, hash)
 
-    def handle_SV_MOVED(self, share_id, node_id, new_share_id, new_parent_id,
+    def _handle_SV_MOVED(self, share_id, node_id, new_share_id, new_parent_id,
                         new_name):
         """on SV_MOVED"""
         key = FSKey(self.m.fs, share_id=share_id, node_id=node_id)
@@ -1129,8 +1005,8 @@
         key = FSKey(self.m.fs, share_id=share_id, node_id=node_id)
         log = FileLogger(self.logger, key)
         ssmr = SyncStateMachineRunner(self.fsm, self.m, key, log)
-        ssmr.check_new_volume_generation(share_id, new_generation)
         ssmr.on_event("AQ_UNLINK_OK", {}, share_id, node_id)
+        ssmr.update_generation(share_id, node_id, new_generation)
 
     def handle_AQ_UNLINK_ERROR(self, share_id, parent_id, node_id, error):
         """on AQ_UNLINK_ERROR"""
@@ -1144,5 +1020,102 @@
         key = FSKey(self.m.fs, share_id=share_id, node_id=node_id)
         log = FileLogger(self.logger, key)
         ssmr = SyncStateMachineRunner(self.fsm, self.m, key, log)
-        ssmr.check_new_volume_generation(share_id, new_generation)
-
+        ssmr.update_generation(share_id, node_id, new_generation)
+
+    def handle_AQ_DELTA_OK(self, volume_id, delta_content, end_generation,
+                           full, free_bytes):
+        """We got a new delta. Apply item by item of the delta.
+
+        Unlike other Sync methods that just defer to Sync State Machine Runner,
+        delta operations sync state in a different way. So this method has a
+        lot of logic that you will not see in the other event handlers.
+        """
+
+        for dt in delta_content:
+            # we only know how to update files
+            try:
+                if not isinstance(dt, delta.FileInfoDelta):
+                    continue
+
+                # we only support files and directories
+                if dt.file_type == delta.DIRECTORY:
+                    is_dir = True
+                elif dt.file_type == delta.FILE:
+                    is_dir = False
+                else:
+                    self.logger.warn("Unknown file type: %r" , dt.file_type)
+                    continue
+
+                # if the node is dead, call the remove handler and forget
+                # about it
+                if not dt.is_live:
+                    self._handle_SV_FILE_DELETED(dt.share_id, dt.node_id)
+                    # nothing else should happen with this
+                    continue
+
+                # here we must call handlers for SV_HASH_NEW, SV_MOVED
+
+                try:
+                    node = self.m.fs.get_by_node_id(dt.share_id, dt.node_id)
+                except KeyError:
+                    # node not there, we must create it
+                    if is_dir:
+                        self._handle_SV_DIR_NEW(dt.share_id, dt.node_id,
+                                               dt.parent_id, dt.name)
+                    else:
+                        self._handle_SV_FILE_NEW(dt.share_id, dt.node_id,
+                                               dt.parent_id, dt.name)
+                    node = self.m.fs.get_by_node_id(dt.share_id, dt.node_id)
+
+                # if the delta is older than the node, skip!
+                if node.generation > dt.generation:
+                    continue
+
+                # if the path changed, we have a move, notify it
+                node_parent_id = self.m.fs.get_by_path(
+                    self.m.fs.get_abspath(node.share_id,
+                                          os.path.dirname(node.path))
+                    ).node_id
+                node_name = os.path.basename(node.path)
+                if dt.parent_id != node_parent_id or dt.name != node_name:
+                    # this was moved, or maybe the server still didn't receive
+                    # the move that happened here
+                    if not self.m.action_q.node_is_with_queued_move(
+                            dt.share_id, dt.node_id):
+                        # signal moved
+                        self._handle_SV_MOVED(
+                            share_id=node.share_id, node_id=node.node_id,
+                            new_share_id=dt.share_id,
+                            new_parent_id=dt.parent_id, new_name=dt.name)
+                    else:
+                        self.logger.info("Not calling _handle_SV_MOVED for "
+                                         "%s:%s due to pending move. "
+                                         "(old parent = %s, new_parent = %s "
+                                         "old_name = %s, new_name = %s)",
+                                         node.share_id, node.node_id,
+                                         node_parent_id, dt.parent_id,
+                                         node_name, dt.name)
+
+
+                # if its a dir, theres nothing else that we do with them except
+                # creating them.
+                # if its a file, we only care about the hash
+                if not is_dir:
+                    self._handle_SV_HASH_NEW(dt.share_id, dt.node_id,
+                                        dt.content_hash)
+
+                # node updated, update generation
+                self.m.fs.set_by_mdid(node.mdid, generation=dt.generation)
+            except Exception:
+                # we trap all exceptions so we can continue processing deltas
+                # even if something fails for one file. We just make sure
+                # we log.
+                self.logger.exception(
+                    "Node delta for %s:%s can't be applied.""",
+                    dt.share_id, dt.node_id)
+
+        self.m.vm.update_free_space(volume_id, free_bytes)
+        self.m.vm.update_generation(volume_id, end_generation)
+
+        if not full:
+            self.m.action_q.get_delta(volume_id, end_generation)

=== modified file 'ubuntuone/syncdaemon/u1fsfsm.ods'
Binary files ubuntuone/syncdaemon/u1fsfsm.ods	2010-05-06 18:07:50 +0000 and ubuntuone/syncdaemon/u1fsfsm.ods	2010-07-22 13:39:48 +0000 differ
=== modified file 'ubuntuone/syncdaemon/u1fsfsm.py'
--- ubuntuone/syncdaemon/u1fsfsm.py	2010-05-06 18:07:50 +0000
+++ ubuntuone/syncdaemon/u1fsfsm.py	2010-07-22 13:39:48 +0000
@@ -704,8 +704,8 @@
                                                      u'has_metadata': u'=',
                                                      u'is_directory': u'='}},
                                       {'ACTION': u'pass',
-                                       'ACTION_FUNC': u'nothing',
-                                       'COMMENTS': u'spurius download finished, ignore',
+                                       'ACTION_FUNC': u'DESPAIR',
+                                       'COMMENTS': u'we dont download directories anymore',
                                        'PARAMETERS': {u'hash_eq_local_hash': u'!NA',
                                                       u'hash_eq_server_hash': u'!NA',
                                                       u'not_authorized': u'NA',
@@ -717,8 +717,8 @@
                                                      u'has_metadata': u'=',
                                                      u'is_directory': u'='}},
                                       {'ACTION': u'merge_from_partial(uuid)',
-                                       'ACTION_FUNC': u'merge_directory',
-                                       'COMMENTS': u'this is the vanilla case, call merge directories',
+                                       'ACTION_FUNC': u'DESPAIR',
+                                       'COMMENTS': u'we dont download directories anymore',
                                        'PARAMETERS': {u'hash_eq_local_hash': u'!NA',
                                                       u'hash_eq_server_hash': u'T',
                                                       u'not_authorized': u'NA',
@@ -730,8 +730,8 @@
                                                      u'has_metadata': u'=',
                                                      u'is_directory': u'='}},
                                       {'ACTION': u'pass',
-                                       'ACTION_FUNC': u'nothing',
-                                       'COMMENTS': u'download we are not interested in',
+                                       'ACTION_FUNC': u'DESPAIR',
+                                       'COMMENTS': u'we dont download directories anymore',
                                        'PARAMETERS': {u'hash_eq_local_hash': u'!NA',
                                                       u'hash_eq_server_hash': u'F',
                                                       u'not_authorized': u'NA',
@@ -2485,8 +2485,8 @@
                                             u'has_metadata': u'T',
                                             u'is_directory': u'='}},
                              {'ACTION': u'pass',
-                              'ACTION_FUNC': u'nothing',
-                              'COMMENTS': u'no news is good news',
+                              'ACTION_FUNC': u'DESPAIR',
+                              'COMMENTS': u'cant set hash on directories',
                               'PARAMETERS': {u'hash_eq_local_hash': u'!NA',
                                              u'hash_eq_server_hash': u'T',
                                              u'not_authorized': u'NA',
@@ -2498,8 +2498,8 @@
                                             u'has_metadata': u'=',
                                             u'is_directory': u'='}},
                              {'ACTION': u'md.set(uuid, server_hash=hash)\npartial = md.create_partial(uuid)\naq.getcontent(*partial)',
-                              'ACTION_FUNC': u'get_dir',
-                              'COMMENTS': u'normal case',
+                              'ACTION_FUNC': u'DESPAIR',
+                              'COMMENTS': u'cant set hash on directories',
                               'PARAMETERS': {u'hash_eq_local_hash': u'!NA',
                                              u'hash_eq_server_hash': u'F',
                                              u'not_authorized': u'NA',
@@ -2511,8 +2511,8 @@
                                             u'has_metadata': u'=',
                                             u'is_directory': u'='}},
                              {'ACTION': u'aq.cancel_download(uuid) \nmd.set(uuid, server_hash=hash)',
-                              'ACTION_FUNC': u'nothing',
-                              'COMMENTS': u'A download for a content object with the same hash is already in progress',
+                              'ACTION_FUNC': u'DESPAIR',
+                              'COMMENTS': u'cant set hash on directories',
                               'PARAMETERS': {u'hash_eq_local_hash': u'!NA',
                                              u'hash_eq_server_hash': u'T',
                                              u'not_authorized': u'NA',
@@ -2524,8 +2524,8 @@
                                             u'has_metadata': u'=',
                                             u'is_directory': u'='}},
                              {'ACTION': u'aq.cancel_download(uuid)\nmd.set(uuid, server_hash=hash)\npartial = md.get_partial(uuid)\naq.getcontent(*partial)',
-                              'ACTION_FUNC': u'reget_dir',
-                              'COMMENTS': u'a download was in progress but the server changed again. Note that this makes it important for AQ_DOWNLOAD_FINISHED to check the server hash.',
+                              'ACTION_FUNC': u'DESPAIR',
+                              'COMMENTS': u'cant set hash on directories',
                               'PARAMETERS': {u'hash_eq_local_hash': u'!NA',
                                              u'hash_eq_server_hash': u'F',
                                              u'not_authorized': u'NA',

=== modified file 'ubuntuone/syncdaemon/volume_manager.py'
--- ubuntuone/syncdaemon/volume_manager.py	2010-07-19 16:03:28 +0000
+++ ubuntuone/syncdaemon/volume_manager.py	2010-07-22 13:39:48 +0000
@@ -110,6 +110,10 @@
                   self.node_id == other.node_id)
         return result
 
+    def __repr__(self):
+        return "<Volume id %r, node_id %r, generation %r>" % (self.volume_id,
+                                                              self.node_id,
+                                                              self.generation)
 
 class Share(Volume):
     """A volume representing a Share."""
@@ -251,7 +255,7 @@
         self.subscribed = subscribed
 
     def __repr__(self):
-        return "<UDF id %r, real path %r>" % (self.id, self.path)
+        return "<UDF id %r, generation %r, real path %r>" % (self.id, self.generation, self.path)
 
     @property
     def ancestors(self):
@@ -393,15 +397,6 @@
             self.m.fs.create(path=root.path,
                              share_id=root.volume_id, is_dir=True)
 
-    def handle_SYS_ROOT_RECEIVED(self, root_id):
-        """Got the root, map it to the  root share."""
-        self.log.debug('handle_SYS_ROOT_RECEIVED(%s)', root_id)
-        self._got_root(root_id)
-        self.m.action_q.inquire_account_info()
-        self.m.action_q.inquire_free_space(request.ROOT)
-        self.refresh_volumes()
-        self.refresh_shares()
-
     def _got_root(self, node_id, free_bytes=None):
         """Set the root node_id to the root share and mdobj."""
         # only set the root if we don't have it
@@ -414,8 +409,12 @@
                 self.m.fs.set_node_id(root.path, node_id)
             root.node_id = node_id
             self.shares[request.ROOT] = root
+            self.m.event_q.push('SYS_ROOT_RECEIVED', root.node_id)
         elif root.node_id != node_id:
             self.m.event_q.push('SYS_ROOT_MISMATCH', root.node_id, node_id)
+        else:
+            # the root node_id match and we already have it
+            self.m.event_q.push('SYS_ROOT_RECEIVED', root.node_id)
 
     def refresh_shares(self):
         """Request the list of shares to the server."""
@@ -439,6 +438,43 @@
                 udfs.append(vol.volume_id)
         self._cleanup_volumes(shares=shares, udfs=udfs)
 
+    def handle_SV_VOLUME_NEW_GENERATION(self, volume_id, generation):
+        """Handle SV_VOLUME_NEW_GENERATION."""
+        self.log.debug('handle_SV_VOLUME_NEW_GENERATION(%r, %r)',
+                       volume_id, generation)
+        volume_id = str(volume_id) # be safe and use a str
+        try:
+            volume = self.get_volume(volume_id)
+        except VolumeDoesNotExist:
+            self.log.warning('Got a SV_VOLUME_NEW_GENERATION for a missing '
+                             'volume: %r with %r', volume_id, generation)
+        else:
+            current_gen = volume.generation
+            if current_gen is None:
+                self.m.action_q.rescan_from_scratch(volume_id)
+            elif current_gen < generation:
+                # XXX: check if we want to impose a hard limit in the size of
+                # the delta and do a rescan from scratch if it's too big
+                self.m.action_q.get_delta(volume_id, current_gen)
+                # TODO/XXX: should we (VM) handle AQ_DELTA_ERROR?
+            elif current_gen >= generation:
+                self.log.info('Got SV_VOLUME_NEW_GENERATION(%r, %r) but volume'
+                              ' is at generation: %r',
+                              volume_id, generation, current_gen)
+
+    def handle_AQ_DELTA_NOT_POSSIBLE(self, volume_id):
+        """Handle AQ_DELTA_NOT_POSSIBLE."""
+        self.log.debug('handle_AQ_DELTA_NOT_POSSIBLE(%r)', volume_id)
+        volume_id = str(volume_id)
+        try:
+            volume = self.get_volume(volume_id)
+        except VolumeDoesNotExist:
+            self.log.warning('Got a AQ_DELTA_NOT_POSSIBLE for a missing '
+                             'volume: %r', volume_id)
+        else:
+            self.log.info('Requesting a rescan from scratch for: %s', volume)
+            self.m.action_q.rescan_from_scratch(volume_id)
+
     def server_rescan(self):
         """Do the 'server rescan'"""
         d = self.m.action_q.query_volumes()
@@ -453,7 +489,7 @@
 
     def _volumes_rescan_cb(self, volumes):
         """Handle the volumes list for server rescan"""
-        self.log.debug('handling volumes rescan')
+        self.log.debug('handling volumes rescan: %r', volumes)
         events = []
         for new_volume in volumes:
             # TODO: all this block might need to be inside a try:except
@@ -465,14 +501,13 @@
                 # A new volume!
                 self.log.debug('New volume! id: %r', volume_id)
                 volume = self._handle_new_volume(new_volume)
-                # we don't have the volume, use the initial generation
-                current_generation = 0
-            else:
-                current_generation = volume.generation or 0 # None case
+            finally:
+                current_generation = volume.generation
             new_generation = new_volume.generation
-            # update the free_bytes on the volume
-            self.update_free_space(volume_id, new_volume.free_bytes)
-            if current_generation < new_generation:
+            self.log.debug('%s current gen: %s vs new gen: %s', volume,
+                           current_generation, new_generation)
+            if current_generation is None or \
+               current_generation < new_generation:
                 # add the event
                 events.append(('SV_VOLUME_NEW_GENERATION',
                                dict(volume_id=volume_id,
@@ -668,31 +703,36 @@
             self.shared[share_id] = share
 
     def handle_AQ_ANSWER_SHARE_OK(self, share_id, answer):
-        """ Handle successfully accepting a share """
+        """Handle successfully accepting a share."""
         if answer == 'Yes':
             share = self.shares[share_id]
+            share.accepted = True
+            self.shares[share.volume_id] = share
             self._create_fsm_object(share.path, share.volume_id, share.node_id)
             self._create_share_dir(share)
-            self.m.action_q.query([(share.volume_id, str(share.node_id), "")])
-            self.m.action_q.inquire_free_space(share.volume_id)
+            # request the delta from the last known generation (should be None)
+            self.m.action_q.get_delta(share.volume_id, share.generation)
 
     def add_share(self, a_share):
         """ Add a share to the share list, and creates the fs mdobj. """
         self.log.info('Adding new share with id: %s - path: %r',
                       a_share.volume_id, a_share.path)
-        # if the share is there, do nothing and return it
         share = self.shares.get(a_share.volume_id)
         is_new_share = share is None
         if share is not None:
             share.accepted = a_share.accepted
             self.shares[share.volume_id] = share
         else:
-            share = self.shares[share.volume_id] = a_share
+            share = a_share
+        if is_new_share or not share.accepted:
+            # if it's a new share or isn't accepted set the generation to None
+            # to force a rescan
+            share.generation = None
+        # store the share
+        self.shares[share.volume_id] = share
         if share.accepted:
             self._create_fsm_object(share.path, share.volume_id, share.node_id)
             self._create_share_dir(share)
-            self.m.action_q.query([(share.volume_id, str(share.node_id), "")])
-            self.m.action_q.inquire_free_space(share.volume_id)
         if is_new_share:
             # push the event only if it's a new share
             self.m.event_q.push('VM_SHARE_CREATED', share.volume_id)
@@ -852,12 +892,15 @@
         """Add the udf to the VM metadata if isn't there.
 
         If it's a new udf, create the directory, hook inotify
-        and execute a query.
+        and request the full delta.
 
         """
         self.log.debug('add_udf: %s', udf)
         if self.udfs.get(udf.volume_id, None) is None:
             self.log.debug('udf not in metadata, adding it!')
+            # if it's a new UDF set the generation to None to force a
+            # rescan
+            udf.generation = None
             self.udfs[udf.volume_id] = udf
             self._create_fsm_object(udf.path, udf.volume_id, udf.node_id)
             # local and server rescan, this will add the inotify hooks
@@ -1063,10 +1106,8 @@
         mdobj = self.m.fs.get_by_path(udf.path)
         d = self.m.lr.scan_dir(mdobj.mdid, udf.path, udfmode=True)
         def server_rescan(_):
-            """Do a query over all known nodes."""
-            data = self.m.fs.get_for_server_rescan_by_path(udf.path)
-            self.m.action_q.query(data)
-            self.m.action_q.inquire_free_space(request.ROOT)
+            """Request the delta from the last known generation."""
+            self.m.action_q.get_delta(udf.id, udf.generation)
         d.addCallback(server_rescan)
         return d
 
@@ -1140,6 +1181,7 @@
 
     def update_generation(self, volume_id, generation):
         """Update the generation of the specified volume."""
+        self.log.debug('update_generation: %r, %r', volume_id, generation)
         vol = self.get_volume(volume_id)
         vol.generation = generation
         if isinstance(vol, (Share, Root)):