Merge lp:~facundo/ubuntuone-client/aq-memory-improv-2 into lp:ubuntuone-client

[Facundo Batista]

Memory improvements in the AQ queues, markers revisited.

Memory improvements in the AQ queues, markers revisited.

A simple one:

- Stop storing a wrapped function in ZipQueue.zip.

And a complicated one:

- Stop storing for ever the markers in DeferredMap.

To achieve this last one improvement, several changes were made, but basically now the responsible of releasing the marker is Sync, and it does it at the same time than setting the node_id in FSM... so now the marker can be removed from the map, as it will not be needed any more.

This implies that the commands must get the marker deferred when they're queued, so I also changed that.

I'll do another branch after this one to fix some method names in the commands ('start', that is actually 'queue', and 'start_unqueued'), and to simplify how the markers are retrieved and demarked in ACQ. This branch is large enough (2k lines, but +1100 of those are tests, and ~300 of spreadsheet).

[Lucio Torre (lucio.torre)]

nice work

review: Approve

[Guillermo Gonzalez (verterok)]

nice

review: Approve

=== modified file 'contrib/testing/testcase.py'
--- contrib/testing/testcase.py	2010-08-25 12:58:01 +0000
+++ contrib/testing/testcase.py	2010-08-30 17:41:19 +0000
@@ -120,14 +120,7 @@
         self.eq = self.event_queue = eq
         self.uploading = {}
         self.downloading = {}
-        # pylint: disable-msg=C0103
-        class UUID_Map(object):
-            """mock uuid map"""
-            def set(self, *args):
-                """mock set method"""
-                pass
-
-        self.uuid_map = UUID_Map()
+        self.uuid_map = action_queue.DeferredMap()
         self.content_queue = action_queue.ContentQueue('CONTENT_QUEUE', self)
         self.meta_queue = action_queue.RequestQueue('META_QUEUE', self)
 

=== modified file 'tests/syncdaemon/test_action_queue.py'
--- tests/syncdaemon/test_action_queue.py	2010-08-23 19:52:04 +0000
+++ tests/syncdaemon/test_action_queue.py	2010-08-30 17:41:19 +0000
@@ -52,7 +52,7 @@
     NoisyRequestQueue, RequestQueue, UploadProgressWrapper, Upload,
     CreateShare, GetPublicFiles, GetDelta, GetDeltaFromScratch,
     TRANSFER_PROGRESS_THRESHOLD, Unlink, Move, MakeFile, MakeDir,
-    DeltaList,
+    DeltaList, ZipQueue, DeferredMap, UniqueRequestQueue
 )
 from ubuntuone.syncdaemon.event_queue import EventQueue, EVENTS
 from ubuntuone.syncdaemon.marker import MDMarker
@@ -90,6 +90,10 @@
     executed = False
     when_executing = 'ok'
 
+    def __init__(self, share_id=None, node_id=None):
+        self.share_id = share_id
+        self.node_id = node_id
+
     def run(self):
         """Run that just succeeds."""
         self.executed = True
@@ -234,23 +238,20 @@
         self.action_queue.upload('foo', 'bar', 0, 0, 0, 0, 0)
         self.assertEqual(len(self.action_queue.content_queue.waiting), 1)
 
+    @defer.inlineCallbacks
     def test_content_queue_has_only_one_op_per_node_even_with_markers(self):
         """ Check that the content queue uniquifies operations per node.
 
         Even when some of the operations were added using markers.
         """
-        self.action_queue.download('foo', 'bar', 0, 0)
+        m1 = MDMarker('foo')
+        m2 = MDMarker('bar')
+        yield self.action_queue.download(m1, m2, 0, 0)
         self.action_queue.uuid_map.set('foo', 'feh')
         self.action_queue.uuid_map.set('bar', 'bah')
         self.action_queue.upload('feh', 'bah', 0, 0, 0, 0, 0)
         self.assertEqual(len(self.action_queue.content_queue.waiting), 1)
 
-    def test_aq_resolve_uuid_maybe(self):
-        """Check action_queue.resolve_uuid_maybe does what it's supposed to."""
-        self.assertEqual(self.action_queue.resolve_uuid_maybe('foo'), 'foo')
-        self.action_queue.uuid_map.set('foo', 'feh')
-        self.assertEqual(self.action_queue.resolve_uuid_maybe('foo'), 'feh')
-
     @defer.inlineCallbacks
     def test_get_root_and_demark(self):
         """Get the received Root and demark its mdid."""
@@ -614,6 +615,216 @@
                                 ])
 
 
+class TestDeferredMap(TwistedTestCase):
+    """Test the deferred map."""
+
+    def setUp(self):
+        """Set up."""
+        self.dm = DeferredMap()
+
+    def test_one_get_returns_stored_deferred(self):
+        """Get will return the stored deferred."""
+        d = self.dm.get('foo')
+        self.assertEqual(self.dm.waiting, {'foo': [d]})
+
+    def test_two_gets_returns_second_deferred_other_key(self):
+        """A second get for other key will return other deferred."""
+        d1 = self.dm.get('foo')
+        d2 = self.dm.get('bar')
+        self.assertEqual(self.dm.waiting, {'foo': [d1], 'bar': [d2]})
+
+    def test_two_gets_returns_second_deferred_same_key(self):
+        """A second get for the same key will return other deferred."""
+        d1 = self.dm.get('foo')
+        d2 = self.dm.get('foo')
+        self.assertEqual(self.dm.waiting, {'foo': [d1, d2]})
+
+    def test_mixed_gets(self):
+        """Several gets with different keys."""
+        d1 = self.dm.get('foo')
+        d2 = self.dm.get('bar')
+        d3 = self.dm.get('foo')
+        d4 = self.dm.get('baz')
+        self.assertEqual(self.dm.waiting,
+                         {'foo': [d1, d3], 'bar': [d2], 'baz': [d4]})
+
+    def test_set_to_nothing(self):
+        """It's ok to set a key that is not being waited."""
+        self.dm.set('not there', 'value')
+
+    @defer.inlineCallbacks
+    def test_set_fires_deferred_single(self):
+        """The set fires the unique waiting deferred with the value."""
+        d1 = self.dm.get('foo')
+        d2 = self.dm.get('bar')
+        d3 = self.dm.get('foo')
+        self.assertEqual(self.dm.waiting, {'foo': [d1, d3], 'bar': [d2]})
+
+        self.dm.set('bar', 'value')
+        res = yield d2
+        self.assertEqual(res, 'value')
+        self.assertEqual(self.dm.waiting, {'foo': [d1, d3]})
+
+    @defer.inlineCallbacks
+    def test_set_fires_deferred_multiple(self):
+        """The set fires the multiple waiting deferreds with the value."""
+        d1 = self.dm.get('foo')
+        d2 = self.dm.get('bar')
+        d3 = self.dm.get('foo')
+        self.assertEqual(self.dm.waiting, {'foo': [d1, d3], 'bar': [d2]})
+
+        self.dm.set('foo', 'value')
+        res1 = yield d1
+        res2 = yield d3
+        self.assertEqual(res1, 'value')
+        self.assertEqual(res2, 'value')
+        self.assertEqual(self.dm.waiting, {'bar': [d2]})
+
+    def test_err_to_nothing(self):
+        """It's ok to err a key that is not being waited."""
+        self.dm.err('not there', 'failure')
+
+    @defer.inlineCallbacks
+    def test_err_fires_deferred_single(self):
+        """The set fires the unique waiting deferred with the failure."""
+        d1 = self.dm.get('foo')
+        d2 = self.dm.get('bar')
+        d3 = self.dm.get('foo')
+        self.assertEqual(self.dm.waiting, {'foo': [d1, d3], 'bar': [d2]})
+
+        exc = Exception('problem!')
+        self.dm.err('bar', Failure(exc))
+        try:
+            yield d2
+        except Exception, e:
+            self.assertEqual(e, exc)
+        else:
+            self.fail("It didn't fired the deferred with a failure!")
+        self.assertEqual(self.dm.waiting, {'foo': [d1, d3]})
+
+
+
+class TestUniqueRequestQueue(TwistedTestCase):
+    """Tests for the UniqueRequestQueue."""
+
+    def setUp(self):
+        """Set up."""
+
+        class FakeAQ(object):
+            """Fake AQ."""
+            event_queue = self.eq = FakedEventQueue()
+
+        self.urq = UniqueRequestQueue(name='META_QUEUE', action_queue=FakeAQ())
+
+        # add a Memento handler to the logger
+        self.handler = MementoHandler()
+        self.handler.setLevel(logging.DEBUG)
+        self._logger = logging.getLogger('ubuntuone.SyncDaemon')
+        self._logger.addHandler(self.handler)
+
+    def tearDown(self):
+        """Tear down."""
+        self._logger.removeHandler(self.handler)
+        self.eq.shutdown()
+
+    def test_queue_one_ok(self):
+        """First command is queued."""
+        cmd = FakeCommand('a', 'b')
+        self.urq.queue(cmd)
+        self.assertEqual(list(self.urq.waiting), [cmd])
+
+    def test_queue_second_ok_if_different(self):
+        """Second command is queued if different than first."""
+        cmd1 = FakeCommand('a', 'b')
+        cmd2 = FakeCommand('a', 'c')
+        self.urq.queue(cmd1)
+        self.urq.queue(cmd2)
+        self.assertEqual(list(self.urq.waiting), [cmd1, cmd2])
+
+    def test_queue_second_equal_first_removes(self):
+        """Second command removes the first one if equal."""
+        cmd1 = FakeCommand('a', 'b')
+        cmd2 = FakeCommand('a', 'b')
+        self.urq.queue(cmd1)
+        self.urq.queue(cmd2)
+        self.assertEqual(list(self.urq.waiting), [cmd2])
+        self.assertTrue(self.handler.check_debug("Request removed"))
+
+    def test_queue_complex(self):
+        """A more complex scenario."""
+        cmd1 = FakeCommand('a', 'b')
+        cmd2 = FakeCommand('a', 'c')
+        cmd3 = FakeCommand('a', 'd')
+        cmd4 = FakeCommand('a', 'e')
+        cmd5 = FakeCommand('a', 'd') # same than cmd3
+        for cmd in cmd1, cmd2, cmd3, cmd4, cmd5:
+            self.urq.queue(cmd)
+        self.assertEqual(list(self.urq.waiting), [cmd1, cmd2, cmd4, cmd5])
+
+
+class TestZipQueue(TwistedTestCase):
+    """Test the zipping queue."""
+
+    def setUp(self):
+        """Set up."""
+        self.zq = ZipQueue()
+
+    @defer.inlineCallbacks
+    def test_zip_calls_compress_in_thread(self):
+        """Test that self._compress is called in another thread."""
+        called = []
+        def fake_compress(deferred, upload):
+            """Fake the _compress method."""
+            self.assertEqual(upload, 'foo')
+            called.append(True)
+            deferred.callback(True)
+
+        self.zq._compress = fake_compress
+        yield self.zq.zip('foo')
+        self.assertTrue(called)
+
+    @defer.inlineCallbacks
+    def test_zip_acquire_lock(self):
+        """Test that it acquires the lock."""
+        called = []
+        self.zq._compress = lambda deferred, upload: deferred.callback(True)
+
+        def fake_acquire():
+            """Fake the acquire method."""
+            self.zq.tokens = 1
+            called.append(True)
+            return defer.succeed(True)
+
+        self.zq.acquire = fake_acquire
+        yield self.zq.zip('foo')
+        self.assertTrue(called)
+
+    @defer.inlineCallbacks
+    def test_zip_release_lock_ok(self):
+        """Test that it releases the lock when all ok."""
+        called = []
+        self.zq._compress = lambda deferred, upload: deferred.callback(True)
+        self.zq.release = lambda: called.append(True)
+
+        yield self.zq.zip('foo')
+        self.assertTrue(called)
+
+    @defer.inlineCallbacks
+    def test_zip_release_lock_error(self):
+        """Test that it releases the lock even on error."""
+        called = []
+        exc = Exception('bad')
+        self.zq._compress = lambda deferred, upload: deferred.errback(exc)
+        self.zq.release = lambda: called.append(True)
+
+        try:
+            yield self.zq.zip('foo')
+        except Exception, e:
+            # need to silent the exception we're generating in the test
+            self.assertEqual(e, exc)
+        self.assertTrue(called)
+
+
 class FactoryBaseTestCase(BasicTestCase):
     """Helper for by-pass Twisted."""
 
@@ -1067,9 +1278,13 @@
             b = 'foo'
             c = u'año'
 
-        request_queue = RequestQueue(name='fu', action_queue=self.action_queue)
+            def _run(self):
+                return defer.succeed(True)
+
+        request_queue = RequestQueue(name='META_QUEUE',
+                                     action_queue=self.action_queue)
         self.cmd = MyCommand(request_queue)
-        self.cmd.start_unqueued()  # just for it to create its logger
+        self.cmd.log = self.cmd.make_logger()
 
         return res
 
@@ -1097,66 +1312,49 @@
         result = yield d
         self.assertEqual(result, ['node_id'])
         self.assertTrue(self.handler.check_debug(
-                        "waiting for the real value of 'foo'"))
+                        "waiting for the real value of marker:foo"))
 
     @defer.inlineCallbacks
     def test_demark_with_marker_ready(self):
         """Test demark with a marker that had data."""
         marker = MDMarker('foo')
+        d = self.cmd.demark(marker)
         self.action_queue.uuid_map.set(marker, 'node_id')
-        result = yield self.cmd.demark(marker)
+        result = yield d
         self.assertEqual(result, ['node_id'])
         self.assertTrue(self.handler.check_debug(
-                        "waiting for the real value of 'foo'"))
+                        "waiting for the real value of marker:foo"))
 
     @defer.inlineCallbacks
     def test_demark_mixed_markers(self):
         """Test demark with both a marker and not."""
         # call demark with both
         marker = MDMarker('foo')
+        d = self.cmd.demark('notamarker', marker)
         self.action_queue.uuid_map.set(marker, 'node_id')
-        result = yield self.cmd.demark('notamarker', marker)
+        result = yield d
 
         # check
         self.assertEqual(result, ['notamarker', 'node_id'])
         self.assertTrue(self.handler.check_debug(
-                        "waiting for the real value of 'foo'"))
+                        "waiting for the real value of marker:foo"))
         self.assertFalse(self.handler.check_debug(
                          "waiting for the real value of 'notamarker'"))
 
     @defer.inlineCallbacks
-    def test_demark_marker_ready_got_ok(self):
-        """Test demark getting a marker triggered ok now."""
-        marker = MDMarker('foo')
-        self.action_queue.uuid_map.set(marker, 'node_id')
-        result = yield self.cmd.demark(marker)
-        self.assertEqual(result, ['node_id'])
-        self.assertTrue(self.handler.check_debug("got 'foo'"))
-
-    @defer.inlineCallbacks
     def test_demark_marker_future_got_ok(self):
         """Test demark getting a marker triggered ok later."""
         # don't have the info now
         marker = MDMarker('foo')
         d = self.cmd.demark(marker)
-        self.assertFalse(self.handler.check_debug("got 'foo'"))
+        self.assertFalse(self.handler.check_debug("for marker:foo"))
 
         # set and check
         self.action_queue.uuid_map.set(marker, 'node_id')
         result = yield d
         self.assertEqual(result, ['node_id'])
-        self.assertTrue(self.handler.check_debug("got 'foo'"))
-
-    @defer.inlineCallbacks
-    def test_demark_marker_ready_got_failure(self):
-        """Test demark getting a marker triggered with failure now."""
-        marker = MDMarker('foo')
-        self.action_queue.uuid_map.err(marker, Failure(Exception('bad')))
-        try:
-            yield self.cmd.demark(marker)
-        except Exception, e:
-            self.assertEqual(str(e), 'bad')
-            self.assertTrue(self.handler.check_error("failed 'foo'"))
+        self.assertTrue(self.handler.check_debug(
+                        "for marker:foo got value 'node_id'"))
 
     @defer.inlineCallbacks
     def test_demark_marker_future_got_failure(self):
@@ -1164,7 +1362,7 @@
         # don't have the info now
         marker = MDMarker('foo')
         d = self.cmd.demark(marker)
-        self.assertFalse(self.handler.check_error("failed 'foo'"))
+        self.assertFalse(self.handler.check_error("failed marker:foo"))
 
         # set the marker and check
         self.action_queue.uuid_map.err(marker, Failure(Exception('bad')))
@@ -1172,8 +1370,113 @@
             yield d
         except Exception, e:
             self.assertEqual(str(e), 'bad')
-            self.assertTrue(self.handler.check_error("failed 'foo'"))
-
+            self.assertTrue(self.handler.check_error("failed marker:foo"))
+
+    @defer.inlineCallbacks
+    def test_run_sets_running(self):
+        """Set the flag."""
+        assert not self.cmd.running
+        d = defer.Deferred()
+        def check():
+            """Checks."""
+            self.assertTrue(self.cmd.running)
+            d.callback(True)
+
+        self.cmd._start = check
+        self.cmd.run()
+        # wait the deferred to be sure it was checked
+        yield d
+
+    @defer.inlineCallbacks
+    def test_run_call_start_if_not_started(self):
+        """Call ._start if it's not started."""
+        called = []
+        self.cmd._start = lambda: called.append(True) or defer.succeed(True)
+        self.cmd.start_done = False
+        self.cmd.markers_resolved_deferred.callback(True)
+        yield self.cmd.run()
+        self.assertTrue(called)
+        self.assertTrue(self.cmd.start_done)
+
+    def test_run_dont_call_start_if_started(self):
+        """Do not call ._start if it's started."""
+        called = []
+        self.cmd._start = lambda: called.append(True) or defer.Deferred()
+        self.cmd.start_done = True
+        self.cmd.run()
+        self.assertFalse(called)
+
+    def test_run_waits_markers_dereferencing(self):
+        """Don't call _ready_to_run until have the markers."""
+        called = []
+        self.cmd._ready_to_run = lambda: called.append(True)
+        self.cmd.run()
+        self.assertFalse(called)
+        self.cmd.markers_resolved_deferred.callback(True)
+        self.assertTrue(called)
+
+    def test_run_not_enderrback_if_ok(self):
+        """If _start is ok, do not call end_errback."""
+        called = []
+        self.cmd.end_errback = lambda _: called.append(True)
+        # default _start will always succeed
+        self.cmd.run()
+        self.assertFalse(called)
+
+    def test_run_enderrback_if_problem(self):
+        """If _start fails, call end_errback."""
+        called = []
+        self.cmd.end_errback = lambda _: called.append(True)
+        self.cmd._start = lambda: defer.fail(Exception('foo'))
+        self.cmd.run()
+        self.assertTrue(called)
+
+    @defer.inlineCallbacks
+    def test_run_done_running_if_ok(self):
+        """If _start is ok, done running."""
+        # default _start will always succeed
+        self.cmd.markers_resolved_deferred.callback(True)
+        yield self.cmd.run()
+        self.assertFalse(self.cmd.running)
+
+    def test_run_done_running_if_problem(self):
+        """If _start fails, done running."""
+        self.cmd.end_errback = lambda _: None # consume the failure
+        self.cmd._start = lambda: defer.fail(Exception('foo'))
+        self.cmd.run()
+        self.assertFalse(self.cmd.running)
+
+    def test_start_calls_unqueued(self):
+        """Do the pre-queue when queued."""
+        called = []
+        self.cmd.start_unqueued = lambda: called.append(True)
+        self.cmd.start()
+        self.assertTrue(called)
+
+    def test_unqueued_calls_demark(self):
+        """Pre-queue implies dereferencing markers."""
+        called = []
+        self.cmd._get_possible_markers = lambda: called.append(True) or ()
+        self.cmd.start_unqueued()
+        self.assertTrue(called)
+
+    @defer.inlineCallbacks
+    def test_start_default(self):
+        """Default _start just returns a triggered deferred and sets done."""
+        yield self.cmd._start()
+
+    @defer.inlineCallbacks
+    def test_get_possible_markers_default(self):
+        """Non-overwritten method just returns a triggered deferred."""
+        yield self.cmd._get_possible_markers()
+
+    @defer.inlineCallbacks
+    def test_marker_storing(self):
+        """Pre-queueing deferres the marker storing."""
+        d = defer.Deferred()
+        self.cmd.store_marker_result = lambda _: d.callback(True)
+        self.cmd.start_unqueued()
+        yield d
 
 class CreateUDFTestCase(ConnectedBaseTestCase):
     """Test for CreateUDF ActionQueueCommand."""
@@ -1588,6 +1891,17 @@
         self.assertIn(event, self.command.action_queue.event_queue.events)
         self.assertTrue(res is None)
 
+    def test_get_possible_markers(self):
+        """Test that it returns the correct values."""
+        res = self.command._get_possible_markers()
+        self.assertEqual(res, (VOLUME, NODE))
+
+    def test_store_marker_result(self):
+        """Test that it stores the correct values."""
+        self.command.store_marker_result(('a', 'b'))
+        self.assertEqual(self.command.share_id, 'a')
+        self.assertEqual(self.command.node_id, 'b')
+
 
 class GetPublicFilesTestCase(ConnectedBaseTestCase):
     """Tests for GetPublicFiles ActionQueueCommand."""
@@ -1783,6 +2097,17 @@
         self.assertEqual(self.command.n_bytes_read_last,
                          5+TRANSFER_PROGRESS_THRESHOLD)
 
+    def test_get_possible_markers(self):
+        """Test that it returns the correct values."""
+        res = self.command._get_possible_markers()
+        self.assertEqual(res, ('a_share_id', 'a_node_id'))
+
+    def test_store_marker_result(self):
+        """Test that it stores the correct values."""
+        self.command.store_marker_result(('a', 'b'))
+        self.assertEqual(self.command.share_id, 'a')
+        self.assertEqual(self.command.node_id, 'b')
+
 
 class UploadUnconnectedTestCase(FactoryBaseTestCase):
     """Test for Upload ActionQueueCommand, no connection"""
@@ -1832,6 +2157,7 @@
         self.command._run()
         self.assertEqual(self.command.n_bytes_written_last, 0)
 
+
 class UploadProgressWrapperTestCase(BaseTwistedTestCase):
     """Test for the UploadProgressWrapper helper class."""
 
@@ -2000,6 +2326,17 @@
         self.assertEqual(self.command.n_bytes_written_last,
                          14+TRANSFER_PROGRESS_THRESHOLD)
 
+    def test_get_possible_markers(self):
+        """Test that it returns the correct values."""
+        res = self.command._get_possible_markers()
+        self.assertEqual(res, (self.command.share_id, 'a_node_id'))
+
+    def test_store_marker_result(self):
+        """Test that it stores the correct values."""
+        self.command.store_marker_result(('a', 'b'))
+        self.assertEqual(self.command.share_id, 'a')
+        self.assertEqual(self.command.node_id, 'b')
+
 
 class CreateShareTestCase(ConnectedBaseTestCase):
     """Test for CreateShare ActionQueueCommand."""
@@ -2060,6 +2397,20 @@
         self.assertEqual('share_name', name)
         self.assertTrue(read_only)
 
+    def test_get_possible_markers(self):
+        """Test that it returns the correct values."""
+        cmd = CreateShare(self.request_queue, 'node_id', '[e-mail]',
+                          'share_name', 'View', 'marker')
+        res = cmd._get_possible_markers()
+        self.assertEqual(res, ('node_id',))
+
+    def test_store_marker_result(self):
+        """Test that it stores the correct values."""
+        cmd = CreateShare(self.request_queue, 'node_id', '[e-mail]',
+                          'share_name', 'View', 'marker')
+        cmd.store_marker_result(('a',))
+        self.assertEqual(cmd.node_id, 'a')
+
 
 class RequestQueueManagerTestCase(FactoryBaseTestCase):
     """Test how RequestQueue manages the queues."""
@@ -3037,6 +3388,20 @@
                     node_id='node_id', new_generation=13)
         self.assertEqual(received, ('AQ_UNLINK_OK', (), info))
 
+    def test_get_possible_markers(self):
+        """Test that it returns the correct values."""
+        cmd = Unlink(self.rq, VOLUME, 'parent_id', 'node_id')
+        res = cmd._get_possible_markers()
+        self.assertEqual(res, (VOLUME, 'node_id', 'parent_id'))
+
+    def test_store_marker_result(self):
+        """Test that it stores the correct values."""
+        cmd = Unlink(self.rq, VOLUME, 'parent_id', 'node_id')
+        cmd.store_marker_result(('a', 'b', 'c'))
+        self.assertEqual(cmd.share_id, 'a')
+        self.assertEqual(cmd.node_id, 'b')
+        self.assertEqual(cmd.parent_id, 'c')
+
 
 class MoveTestCase(ConnectedBaseTestCase):
     """Test for Move ActionQueueCommand."""
@@ -3064,6 +3429,21 @@
         info = dict(share_id=VOLUME, node_id='node', new_generation=13)
         self.assertEqual(received, ('AQ_MOVE_OK', (), info))
 
+    def test_get_possible_markers(self):
+        """Test that it returns the correct values."""
+        cmd = Move(self.rq, VOLUME, 'node', 'o_parent', 'n_parent', 'n_name')
+        res = cmd._get_possible_markers()
+        self.assertEqual(res, (VOLUME, 'node', 'o_parent', 'n_parent'))
+
+    def test_store_marker_result(self):
+        """Test that it stores the correct values."""
+        cmd = Move(self.rq, VOLUME, 'node', 'o_parent', 'n_parent', 'n_name')
+        cmd.store_marker_result(('a', 'b', 'c', 'd'))
+        self.assertEqual(cmd.share_id, 'a')
+        self.assertEqual(cmd.node_id, 'b')
+        self.assertEqual(cmd.old_parent_id, 'c')
+        self.assertEqual(cmd.new_parent_id, 'd')
+
 
 class MakeFileTestCase(ConnectedBaseTestCase):
     """Test for MakeFile ActionQueueCommand."""
@@ -3093,6 +3473,37 @@
                     volume_id=VOLUME)
         self.assertEqual(received, ('AQ_FILE_NEW_OK', (), info))
 
+    def test_handle_failure_push_event(self):
+        """Test AQ_FILE_NEW_ERROR is pushed on error."""
+        # create a request and fill it with succesful information
+        request = client.MakeFile(self.action_queue.client, VOLUME,
+                                  'parent', 'name')
+        request.new_id = 'new_id'
+        request.new_generation = 13
+
+        # create a command and trigger it fail
+        cmd = MakeFile(self.rq, VOLUME, 'parent', 'name', 'marker')
+        failure = Failure(Exception('foo'))
+        cmd.handle_failure(failure)
+
+        # check for successful event
+        received = self.action_queue.event_queue.events[0]
+        info = dict(marker='marker', failure=failure)
+        self.assertEqual(received, ('AQ_FILE_NEW_ERROR', (), info))
+
+    def test_get_possible_markers(self):
+        """Test that it returns the correct values."""
+        cmd = MakeFile(self.rq, VOLUME, 'parent', 'name', 'marker')
+        res = cmd._get_possible_markers()
+        self.assertEqual(res, (VOLUME, 'parent'))
+
+    def test_store_marker_result(self):
+        """Test that it stores the correct values."""
+        cmd = MakeFile(self.rq, VOLUME, 'parent', 'name', 'marker')
+        cmd.store_marker_result(('a', 'b'))
+        self.assertEqual(cmd.share_id, 'a')
+        self.assertEqual(cmd.parent_id, 'b')
+
 
 class MakeDirTestCase(ConnectedBaseTestCase):
     """Test for MakeDir ActionQueueCommand."""
@@ -3122,6 +3533,37 @@
                     volume_id=VOLUME)
         self.assertEqual(received, ('AQ_DIR_NEW_OK', (), info))
 
+    def test_handle_failure_push_event(self):
+        """Test AQ_DIR_NEW_ERROR is pushed on error."""
+        # create a request and fill it with succesful information
+        request = client.MakeDir(self.action_queue.client, VOLUME,
+                                 'parent', 'name')
+        request.new_id = 'new_id'
+        request.new_generation = 13
+
+        # create a command and trigger it fail
+        cmd = MakeDir(self.rq, VOLUME, 'parent', 'name', 'marker')
+        failure = Failure(Exception('foo'))
+        cmd.handle_failure(failure)
+
+        # check for successful event
+        received = self.action_queue.event_queue.events[0]
+        info = dict(marker='marker', failure=failure)
+        self.assertEqual(received, ('AQ_DIR_NEW_ERROR', (), info))
+
+    def test_get_possible_markers(self):
+        """Test that it returns the correct values."""
+        cmd = MakeDir(self.rq, VOLUME, 'parent', 'name', 'marker')
+        res = cmd._get_possible_markers()
+        self.assertEqual(res, (VOLUME, 'parent'))
+
+    def test_store_marker_result(self):
+        """Test that it stores the correct values."""
+        cmd = MakeDir(self.rq, VOLUME, 'parent', 'name', 'marker')
+        cmd.store_marker_result(('a', 'b'))
+        self.assertEqual(cmd.share_id, 'a')
+        self.assertEqual(cmd.parent_id, 'b')
+
 
 class TestDeltaList(unittest.TestCase):
     """Tests for DeltaList."""

=== modified file 'tests/syncdaemon/test_sync.py'
--- tests/syncdaemon/test_sync.py	2010-08-10 14:08:40 +0000
+++ tests/syncdaemon/test_sync.py	2010-08-30 17:41:19 +0000
@@ -27,6 +27,9 @@
 import unittest
 import uuid
 
+from twisted.internet import defer
+from twisted.python.failure import Failure
+
 from contrib.testing.testcase import (
     FakeMain,
     FakeVolumeManager,
@@ -222,6 +225,12 @@
 class TestSync(BaseSync):
     """Test for Sync."""
 
+    def setUp(self):
+        """Set up."""
+        BaseSync.setUp(self)
+        self.sync = Sync(main=self.main)
+        self.fsm = self.main.fs
+
     def test_deleting_open_files_is_no_cause_for_despair(self):
         """test_deleting_open_files_is_no_cause_for_despair."""
         def cb(_):
@@ -254,7 +263,6 @@
 
     def test_handle_AQ_DOWNLOAD_DOES_NOT_EXIST(self):
         """handle_AQ_DOWNLOAD_DOES_NOT_EXIST."""
-        sync = Sync(main=self.main)
         self.called = False
 
         def faked_nothing(ssmr, event, params, *args):
@@ -263,12 +271,11 @@
         self.patch(SyncStateMachineRunner, 'nothing', faked_nothing)
 
         kwargs = dict(share_id='share_id', node_id='node_id')
-        sync.handle_AQ_DOWNLOAD_DOES_NOT_EXIST(**kwargs)
+        self.sync.handle_AQ_DOWNLOAD_DOES_NOT_EXIST(**kwargs)
         self.assertTrue(self.called, 'nothing was called')
 
     def test_handle_FILE_CREATE_while_LOCAL(self):
         """A FS_FILE_CREATE is received with the node in LOCAL."""
-        sync = Sync(main=self.main)
         self.called = False
 
         def faked_nothing(ssmr, event, params, *args):
@@ -277,18 +284,16 @@
         self.patch(SyncStateMachineRunner, 'nothing', faked_nothing)
 
         # create a file and put it in local
-        fsm = self.main.fs
         somepath = os.path.join(self.root, 'somepath')
-        mdid = fsm.create(somepath, '')
-        fsm.set_by_mdid(mdid, local_hash='somehash')
+        mdid = self.fsm.create(somepath, '')
+        self.fsm.set_by_mdid(mdid, local_hash='somehash')
 
         # send the event, and check that it called its .nothing()
-        sync.handle_FS_FILE_CREATE(somepath)
+        self.sync.handle_FS_FILE_CREATE(somepath)
         self.assertTrue(self.called)
 
     def test_SV_HASH_NEW_with_file_uploadinterrupted(self):
         """A SV_HASH_NEW is received after upload interrupted."""
-        sync = Sync(main=self.main)
         self.called = False
 
         def fake_meth(_, event, params, hash):
@@ -300,17 +305,232 @@
 
         # create a file and put it in local, without server_hash, as
         # if the upload was cut in the middle after the make file
-        fsm = self.main.fs
         somepath = os.path.join(self.root, 'somepath')
-        mdid = fsm.create(somepath, '', node_id='node_id')
-        fsm.set_by_mdid(mdid, local_hash='somehash', crc32='crc32',
-                        stat='stat', size='size')
+        mdid = self.fsm.create(somepath, '', node_id='node_id')
+        self.fsm.set_by_mdid(mdid, local_hash='somehash', crc32='crc32',
+                             stat='stat', size='size')
 
-        # send the event and check
-        mdobj = fsm.get_by_mdid(mdid)
-        sync._handle_SV_HASH_NEW(mdobj.share_id, mdobj.node_id, '') # no content
+        # send the event with no content and check
+        mdobj = self.fsm.get_by_mdid(mdid)
+        self.sync._handle_SV_HASH_NEW(mdobj.share_id, mdobj.node_id, '')
         self.assertTrue(self.called)
 
+    def test_AQ_FILE_NEW_OK_with_md_in_none(self):
+        """Created the file, and MD says it's in NONE."""
+        # fake method
+        called = []
+        self.patch(SyncStateMachineRunner, 'new_local_file_created',
+                   lambda *a: called.extend(a))
+
+        # create the node and set it up
+        somepath = os.path.join(self.root, 'somepath')
+        mdid = self.fsm.create(somepath, '', node_id='node_id')
+        assert self.fsm.changed(mdid=mdid) == 'NONE', 'test badly set up'
+
+        # send the event and check args after the ssmr instance
+        mdobj = self.fsm.get_by_mdid(mdid)
+        self.sync.handle_AQ_FILE_NEW_OK(mdobj.share_id, mdid, 'new_id', 'gen')
+        self.assertEqual(called[1:], ['AQ_FILE_NEW_OK', {}, 'new_id', mdid])
+
+    def test_AQ_FILE_NEW_OK_with_md_in_local(self):
+        """Created the file, and MD says it's in LOCAL."""
+        # fake method
+        called = []
+        self.patch(SyncStateMachineRunner, 'new_local_file_created',
+                   lambda *a: called.extend(a))
+
+        # create the node and set it up
+        somepath = os.path.join(self.root, 'somepath')
+        mdid = self.fsm.create(somepath, '', node_id='node_id')
+        self.fsm.set_by_mdid(mdid, local_hash='somehash')
+        assert self.fsm.changed(mdid=mdid) == 'LOCAL', 'test badly set up'
+
+        # send the event and check args after the ssmr instance
+        mdobj = self.fsm.get_by_mdid(mdid)
+        self.sync.handle_AQ_FILE_NEW_OK(mdobj.share_id, mdid, 'new_id', 'gen')
+        self.assertEqual(called[1:], ['AQ_FILE_NEW_OK', {}, 'new_id', mdid])
+
+    def test_AQ_FILE_NEW_OK_no_md(self):
+        """Created the file, but MD is no longer there."""
+        # fake method
+        called = []
+        self.patch(SyncStateMachineRunner, 'release_marker_ok',
+                   lambda *a: called.extend(a))
+
+        # send the event and check args after the ssmr instance
+        self.sync.handle_AQ_FILE_NEW_OK('share', 'mrker', 'new_id', 'gen')
+        self.assertEqual(called[1:], ['AQ_FILE_NEW_OK', {}, 'new_id', 'mrker'])
+
+    def test_AQ_FILE_NEW_OK_md_says_dir(self):
+        """Created the file, but MD says it's now a directory."""
+        # fake method
+        called = []
+        self.patch(SyncStateMachineRunner, 'release_marker_ok',
+                   lambda *a: called.extend(a))
+
+        # create the node as a dir
+        somepath = os.path.join(self.root, 'somepath')
+        mdid = self.fsm.create(somepath, '', node_id='node_id', is_dir=True)
+
+        # send the event and check args after the ssmr instance
+        mdobj = self.fsm.get_by_mdid(mdid)
+        self.sync.handle_AQ_FILE_NEW_OK(mdobj.share_id, mdid, 'new_id', 'gen')
+        self.assertEqual(called[1:], ['AQ_FILE_NEW_OK', {}, 'new_id', mdid])
+
+    def test_AQ_DIR_NEW_OK_md_says_file(self):
+        """Created the dir, but MD says it's now a file."""
+        # fake method
+        called = []
+        self.patch(SyncStateMachineRunner, 'release_marker_ok',
+                   lambda *a: called.extend(a))
+
+        # create the node as a file
+        somepath = os.path.join(self.root, 'somepath')
+        mdid = self.fsm.create(somepath, '', node_id='node_id')
+
+        # send the event and check args after the ssmr instance
+        mdobj = self.fsm.get_by_mdid(mdid)
+        self.sync.handle_AQ_DIR_NEW_OK(mdobj.share_id, mdid, 'new_id', 'gen')
+        self.assertEqual(called[1:], ['AQ_DIR_NEW_OK', {}, 'new_id', mdid])
+
+    def test_AQ_DIR_NEW_OK_no_md(self):
+        """Created the dir, but MD is no longer there."""
+        # fake method
+        called = []
+        self.patch(SyncStateMachineRunner, 'release_marker_ok',
+                   lambda *a: called.extend(a))
+
+        # send the event and check args after the ssmr instance
+        self.sync.handle_AQ_DIR_NEW_OK('share', 'marker', 'new_id', 'gen')
+        self.assertEqual(called[1:], ['AQ_DIR_NEW_OK', {}, 'new_id', 'marker'])
+
+    def test_AQ_DIR_NEW_OK_md_in_NONE(self):
+        """Created the dir, and MD says it's in NONE."""
+        # fake method
+        called = []
+        self.patch(SyncStateMachineRunner, 'new_local_dir_created',
+                   lambda *a: called.extend(a))
+
+        # create the node and set it up
+        somepath = os.path.join(self.root, 'somepath')
+        mdid = self.fsm.create(somepath, '', node_id='node_id', is_dir=True)
+        assert self.fsm.changed(mdid=mdid) == 'NONE', 'test badly set up'
+
+        # send the event and check args after the ssmr instance
+        mdobj = self.fsm.get_by_mdid(mdid)
+        self.sync.handle_AQ_DIR_NEW_OK(mdobj.share_id, mdid, 'new_id', 'gen')
+        self.assertEqual(called[1:], ['AQ_DIR_NEW_OK', {}, 'new_id', mdid])
+
+    def test_AQ_FILE_NEW_ERROR_no_md(self):
+        """Error creating the file, MD is no longer there."""
+        # fake method
+        called = []
+        self.patch(SyncStateMachineRunner, 'release_marker_error',
+                   lambda *a: called.extend(a))
+
+        # send the event and check args after the ssmr instance
+        failure = Failure(Exception('foo'))
+        params = {'not_authorized': 'F', 'not_available': 'F'}
+        self.sync.handle_AQ_FILE_NEW_ERROR('marker', failure)
+        self.assertEqual(called[1:],
+                        ['AQ_FILE_NEW_ERROR', params, failure, 'marker'])
+
+    def test_AQ_FILE_NEW_ERROR_md_ok(self):
+        """Error creating the file, MD is ok."""
+        # fake method
+        called = []
+        realf = SyncStateMachineRunner.filedir_error_in_creation
+        def fake(*args):
+            """Call the original function, but storing the args."""
+            called.extend(args)
+            realf(*args)
+            SyncStateMachineRunner.filedir_error_in_creation = realf
+        SyncStateMachineRunner.filedir_error_in_creation = fake
+
+        # create the node
+        somepath = os.path.join(self.root, 'somepath')
+        mdid = self.fsm.create(somepath, '', node_id='node_id')
+
+        # send the event and check args after the ssmr instance
+        failure = Failure(Exception('foo'))
+        params = {'not_authorized': 'F', 'not_available': 'F'}
+        self.sync.handle_AQ_FILE_NEW_ERROR(mdid, failure)
+        self.assertEqual(called[1:],
+                        ['AQ_FILE_NEW_ERROR', params, failure, mdid])
+
+    def test_AQ_FILE_NEW_ERROR_md_says_dir(self):
+        """Error creating the file, MD says it's now a dir."""
+        # fake method
+        called = []
+        self.patch(SyncStateMachineRunner, 'release_marker_error',
+                   lambda *a: called.extend(a))
+
+        # create the node as a dir
+        somepath = os.path.join(self.root, 'somepath')
+        self.fsm.create(somepath, '', node_id='node_id', is_dir=True)
+
+        # send the event and check args after the ssmr instance
+        failure = Failure(Exception('foo'))
+        params = {'not_authorized': 'F', 'not_available': 'F'}
+        self.sync.handle_AQ_FILE_NEW_ERROR('marker', failure)
+        self.assertEqual(called[1:],
+                        ['AQ_FILE_NEW_ERROR', params, failure, 'marker'])
+
+    def test_AQ_DIR_NEW_ERROR_no_md(self):
+        """Error creating the dir, MD is no longer there."""
+        # fake method
+        called = []
+        self.patch(SyncStateMachineRunner, 'release_marker_error',
+                   lambda *a: called.extend(a))
+
+        # send the event and check args after the ssmr instance
+        failure = Failure(Exception('foo'))
+        params = {'not_authorized': 'F', 'not_available': 'F'}
+        self.sync.handle_AQ_DIR_NEW_ERROR('marker', failure)
+        self.assertEqual(called[1:],
+                        ['AQ_DIR_NEW_ERROR', params, failure, 'marker'])
+
+    def test_AQ_DIR_NEW_ERROR_md_ok(self):
+        """Error creating the dir, MD is ok."""
+        # fake method
+        called = []
+        realf = SyncStateMachineRunner.filedir_error_in_creation
+        def fake(*args):
+            """Call the original function, but storing the args."""
+            called.extend(args)
+            realf(*args)
+            SyncStateMachineRunner.filedir_error_in_creation = realf
+        SyncStateMachineRunner.filedir_error_in_creation = fake
+
+        # create the node
+        somepath = os.path.join(self.root, 'somepath')
+        mdid = self.fsm.create(somepath, '', node_id='node_id', is_dir=True)
+
+        # send the event and check args after the ssmr instance
+        failure = Failure(Exception('foo'))
+        params = {'not_authorized': 'F', 'not_available': 'F'}
+        self.sync.handle_AQ_DIR_NEW_ERROR(mdid, failure)
+        self.assertEqual(called[1:],
+                        ['AQ_DIR_NEW_ERROR', params, failure, mdid])
+
+    def test_AQ_DIR_NEW_ERROR_md_says_file(self):
+        """Error creating the dir, MD says it's now a file."""
+        # fake method
+        called = []
+        self.patch(SyncStateMachineRunner, 'release_marker_error',
+                   lambda *a: called.extend(a))
+
+        # create the node as a file
+        somepath = os.path.join(self.root, 'somepath')
+        self.fsm.create(somepath, '', node_id='node_id')
+
+        # send the event and check args after the ssmr instance
+        failure = Failure(Exception('foo'))
+        params = {'not_authorized': 'F', 'not_available': 'F'}
+        self.sync.handle_AQ_DIR_NEW_ERROR('marker', failure)
+        self.assertEqual(called[1:],
+                        ['AQ_DIR_NEW_ERROR', params, failure, 'marker'])
+
 
 class SyncStateMachineRunnerTestCase(BaseSync):
     """Tests for the SyncStateMachineRunner."""
@@ -318,16 +538,16 @@
     def setUp(self):
         """Init."""
         BaseSync.setUp(self)
+        self.fsm = self.main.fs
+        self.aq = self.main.action_q
 
         # create a file
-        self.fsm = self.main.fs
         somepath = os.path.join(self.root, 'somepath')
         self.mdid = self.fsm.create(somepath, '', node_id='node_id')
 
         key = FSKey(self.main.fs, share_id='', node_id='node_id')
         self.ssmr = SyncStateMachineRunner(fsm=self.main.fs, main=self.main,
                                            key=key, logger=None)
-        self.root = self.mktemp('root')
 
         # log config
         self.handler = MementoHandler()
@@ -405,6 +625,116 @@
         mdobj = self.fsm.get_by_mdid(self.mdid)
         self.assertTrue(mdobj.info.is_partial)
 
+    @defer.inlineCallbacks
+    def test_new_local_file_created(self):
+        """Set the node_id in FSM, and release ok the marker in DeferredMap."""
+        # set up FSM and the DeferredMap
+        somepath = os.path.join(self.root, 'foo')
+        mdid = self.fsm.create(somepath, '')
+        map_d = self.aq.uuid_map.get('marker')
+
+        # create context and call
+        key = FSKey(self.main.fs, path=somepath)
+        ssmr = SyncStateMachineRunner(fsm=self.fsm, main=self.main,
+                                      key=key, logger=None)
+        ssmr.new_local_file_created('some event', {}, 'new_id', 'marker')
+
+        # check
+        mdobj = self.fsm.get_by_mdid(mdid)
+        self.assertEqual(mdobj.node_id, 'new_id')
+        result = yield map_d
+        self.assertEqual(result, 'new_id')
+
+    @defer.inlineCallbacks
+    def test_new_local_dir_created(self):
+        """Set the node_id in FSM, and release ok the marker in DeferredMap."""
+        # set up FSM and the DeferredMap
+        somepath = os.path.join(self.root, 'foo')
+        mdid = self.fsm.create(somepath, '', is_dir=True)
+        map_d = self.aq.uuid_map.get('marker')
+
+        # create context and call
+        key = FSKey(self.main.fs, path=somepath)
+        ssmr = SyncStateMachineRunner(fsm=self.fsm, main=self.main,
+                                      key=key, logger=None)
+        ssmr.new_local_dir_created('some event', {}, 'new_id', 'marker')
+
+        # check
+        mdobj = self.fsm.get_by_mdid(mdid)
+        self.assertEqual(mdobj.node_id, 'new_id')
+        result = yield map_d
+        self.assertEqual(result, 'new_id')
+
+    @defer.inlineCallbacks
+    def test_release_marker_ok(self):
+        """Just release the marker ok in DeferredMap."""
+        # set up the DeferredMap
+        map_d = self.aq.uuid_map.get('marker')
+
+        # create context and call
+        key = FSKey(self.main.fs)
+        ssmr = SyncStateMachineRunner(fsm=self.fsm, main=self.main,
+                                      key=key, logger=None)
+        ssmr.release_marker_ok('some event', {}, 'new_id', 'marker')
+
+        # check
+        result = yield map_d
+        self.assertEqual(result, 'new_id')
+
+    @defer.inlineCallbacks
+    def test_filedir_error_in_creation(self):
+        """Conflict and delete metada, and release the marker with error."""
+        # set up FSM and the DeferredMap
+        somepath = os.path.join(self.root, 'foo')
+        mdid = self.fsm.create(somepath, '')
+        map_d = self.aq.uuid_map.get('mrker')
+
+        # patch to control the call to key
+        called = []
+        self.fsm.move_to_conflict = lambda m: called.append(m)
+        self.fsm.delete_metadata = lambda p: called.append(p)
+
+        # create context and call
+        key = FSKey(self.main.fs, path=somepath)
+        ssmr = SyncStateMachineRunner(fsm=self.fsm, main=self.main,
+                                      key=key, logger=None)
+        exc = Exception('foo')
+        ssmr.filedir_error_in_creation('some event', {}, Failure(exc), 'mrker')
+
+        # check
+        self.assertEqual(called, [mdid, somepath])
+        try:
+            yield map_d
+        except Exception, e:
+            # silence the received exception
+            self.assertEqual(e, exc)
+        else:
+            # no exception? fail!!
+            self.fail("The marker was released without failure!")
+
+    @defer.inlineCallbacks
+    def test_release_marker_error(self):
+        """Just release the marker with failure in DeferredMap."""
+        # set up the DeferredMap
+        map_d = self.aq.uuid_map.get('mrker')
+
+        # create context and call
+        key = FSKey(self.main.fs)
+        ssmr = SyncStateMachineRunner(fsm=self.fsm, main=self.main,
+                                      key=key, logger=None)
+        exc = Exception('foo')
+        ssmr.release_marker_error('some event', {}, Failure(exc), 'mrker')
+
+        # check
+        try:
+            yield map_d
+        except Exception, e:
+            # silence the received exception
+            self.assertEqual(e, exc)
+        else:
+            # no exception? fail!!
+            self.fail("The marker was released without failure!")
+
 
 class FakedState(object):
     """A faked state."""

=== modified file 'ubuntuone/syncdaemon/action_queue.py'
--- ubuntuone/syncdaemon/action_queue.py	2010-08-23 19:52:04 +0000
+++ ubuntuone/syncdaemon/action_queue.py	2010-08-30 17:41:19 +0000
@@ -209,9 +209,7 @@
         self.tokens = self.limit = 10
 
     def acquire(self):
-        """
-        return a deferred which fires on token acquisition.
-        """
+        """Return a deferred which fires on token acquisition."""
         assert self.tokens >= 0, "Tokens should never be negative"
         d = defer.Deferred()
         if not self.tokens:
@@ -222,8 +220,7 @@
         return d
 
     def release(self):
-        """
-        Release the token.
+        """Release the token.
 
         Should be called by whoever did the acquire() when the shared
         resource is free.
@@ -283,18 +280,20 @@
         else:
             reactor.callFromThread(deferred.callback, True)
 
+    @defer.inlineCallbacks
     def zip(self, upload):
-        """
-        Acquire, do the compression in a thread, release.
-        """
-        d_zip = defer.Deferred()
-        d_lck = self.acquire()
-        d_lck.addCallback(
-            lambda _: reactor.callInThread(self._compress,
-                                           d_zip, upload) or d_zip)
-        d_lck.addBoth(passit(lambda _: self.release()))
-
-        return d_lck
+        """Acquire, do the compression in a thread, release."""
+        deferred = defer.Deferred()
+
+        yield self.acquire()
+        try:
+            reactor.callInThread(self._compress, deferred, upload)
+        finally:
+            self.release()
+
+        # let's wait _compress to finish
+        yield deferred
+
 
 class RequestQueue(object):
     """
@@ -442,23 +441,23 @@
 
 
 class UniqueRequestQueue(RequestQueue):
-    """A RequestQueue.
+    """A unique RequestQueue.
 
     It only ever queues one command for each (share_id, node_id) pair.
-
     """
 
     def queue(self, command, on_top=False):
         """Add a command to the queue.
 
-        If there are commands in the queue for the
-        same node, they will be dropped on the floor and laughed at.
+        If there is a command in the queue for the same node, it (the old
+        one) will be removed, leaving queued only the new one.
         """
-        for wc in iter(self.waiting):
-            wc_share = self.action_queue.resolve_uuid_maybe(wc.share_id)
-            wc_node = self.action_queue.resolve_uuid_maybe(wc.node_id)
-            if wc_share == command.share_id and wc_node == command.node_id:
+        for wc in self.waiting:
+            if wc.share_id == command.share_id and \
+               wc.node_id == command.node_id:
                 self.waiting.remove(wc)
+                m = "Request removed (%s), queing other (%s) for same node."
+                logger.debug(m, wc, command)
                 break
         return super(UniqueRequestQueue, self).queue(command, on_top)
 
@@ -506,31 +505,16 @@
 
 
 class DeferredMap(object):
-    """A mapping of deferred values. Or a deferred map of values.
-
-    Or a mapping that returns deferreds and then fires them when it has
-    the value.
-
+    """A mapping of deferred values.
+
+    Return deferreds for a key that are fired (succesfully or not) later.
     """
 
     def __init__(self):
         self.waiting = defaultdict(list)
-        self.failed = {}
-        self.map = {}
 
     def get(self, key):
-        """Get the value for the given key.
-
-        This always returns a deferred; when we already know the value
-        we return a `succeed`, and if we don't know the value because
-        it failed we return a `fail`; otherwise we return a plain
-        unfired `Deferred`, and add it to the list of deferreds to
-        call when we actually get the value.
-        """
-        if key in self.map:
-            return defer.succeed(self.map[key])
-        if key in self.failed:
-            return defer.fail(Exception(self.failed[key]))
+        """Return a deferred for the given key."""
         d = defer.Deferred()
         self.waiting[key].append(d)
         return d
@@ -538,31 +522,22 @@
     def set(self, key, value):
         """We've got the value for a key!
 
-        Write it down in the map, and fire the waiting deferreds.
+        If it was waited, fire the waiting deferreds and remove the key.
         """
-        if key not in self.map:
-            self.map[key] = value
-            for d in self.waiting.pop(key, ()):
+        if key in self.waiting:
+            deferreds = self.waiting.pop(key)
+            for d in deferreds:
                 d.callback(value)
-        elif self.map[key] != value:
-            if key in self.map:
-                raise KeyError("key is taken -- dunno what to do")
 
     def err(self, key, failure):
         """Something went terribly wrong in the process of getting a value.
 
-        Break the news to the waiting deferreds.
-        """
-        self.failed[key] = failure.getErrorMessage()
-        for d in self.waiting.pop(key, ()):
-            d.errback(failure)
-
-    def resolve_maybe(self, key):
-        """Return either the mapping of key, or key itself.
-
-        The former if key has been resolved, the later if not.
-        """
-        return self.map.get(key, key)
+        Break the news to the waiting deferreds and remove the key.
+        """
+        if key in self.waiting:
+            deferreds = self.waiting.pop(key)
+            for d in deferreds:
+                d.errback(failure)
 
 
 class UploadProgressWrapper(object):
@@ -1001,23 +976,17 @@
         defer.returnValue(result.volumes)
 
     def make_file(self, share_id, parent_id, name, marker):
-        """
-        See .interfaces.IMetaQueue
-        """
+        """See .interfaces.IMetaQueue."""
         return MakeFile(self.meta_queue, share_id, parent_id,
                         name, marker).start()
 
     def make_dir(self, share_id, parent_id, name, marker):
-        """
-        See .interfaces.IMetaQueue
-        """
+        """See .interfaces.IMetaQueue."""
         return MakeDir(self.meta_queue, share_id, parent_id,
                        name, marker).start()
 
     def move(self, share_id, node_id, old_parent_id, new_parent_id, new_name):
-        """
-        See .interfaces.IMetaQueue
-        """
+        """See .interfaces.IMetaQueue."""
         return Move(self.meta_queue, share_id, node_id, old_parent_id,
                     new_parent_id, new_name).start()
 
@@ -1147,14 +1116,6 @@
         """True if a Move is queued for that node."""
         return self.meta_queue.node_is_queued(Move, share_id, node_id)
 
-    def resolve_uuid_maybe(self, marker_maybe):
-        """Resolve something tha may be a marker.
-
-        Resolve the maybe_marker if it is a marker and has been
-        resolved. Otherwise just return the marker_maybe back again.
-        """
-        return self.uuid_map.resolve_maybe(marker_maybe)
-
     def get_delta(self, volume_id, generation):
         """Get a delta from generation for the volume.
 
@@ -1205,7 +1166,8 @@
 
     logged_attrs = ()
 
-    __slots__ = ('_queue', 'start_done', 'running', 'log')
+    __slots__ = ('_queue', 'start_done', 'running', 'log',
+                 'markers_resolved_deferred')
 
     def __init__(self, request_queue):
         """Initialize a command instance."""
@@ -1213,6 +1175,7 @@
         self.start_done = False
         self.running = False
         self.log = None
+        self.markers_resolved_deferred = defer.Deferred()
 
     def to_dict(self):
         """Dump logged attributes to a dict."""
@@ -1240,20 +1203,32 @@
     @defer.inlineCallbacks
     def demark(self, *maybe_markers):
         """Arrange to have maybe_markers realized."""
-        results = []
+        # we need to issue all the DeferredMap.get's right now, to be
+        # dereferenced later
+        waiting_structure = []
         for marker in maybe_markers:
             if IMarker.providedBy(marker):
-                self.log.debug("waiting for the real value of '%s'", marker)
+                self.log.debug("waiting for the real value of %r", marker)
+                d = self.action_queue.uuid_map.get(marker)
+                waiting_structure.append((marker, True, d))
+            else:
+                waiting_structure.append((marker, False, None))
+
+        # now, we wait for all the dereferencings...
+        results = []
+        for (marker, need_to_wait, deferred) in waiting_structure:
+            if need_to_wait:
                 try:
-                    result = yield self.action_queue.uuid_map.get(marker)
+                    value = yield deferred
                 except:
-                    self.log.error("failed '%s'", marker)
+                    self.log.error("failed %r", marker)
                     raise
                 else:
-                    self.log.debug("got '%s'", marker)
-                    results.append(result)
+                    self.log.debug("for %r got value %r", marker, value)
+                    results.append(value)
             else:
                 results.append(marker)
+
         defer.returnValue(results)
 
     def end_callback(self, arg):
@@ -1280,9 +1255,18 @@
         else:
             return self.handle_failure(failure)
 
+    @defer.inlineCallbacks
     def start_unqueued(self):
         """Do basic pre-start setup."""
         self.log = self.make_logger()
+        maybe_markers = self._get_possible_markers()
+        result = yield self.demark(*maybe_markers)
+        self.store_marker_result(result)
+        self.markers_resolved_deferred.callback(True)
+
+    def _get_possible_markers(self):
+        """Nothing if not overwritten."""
+        return ()
 
     def start(self, _=None):
         """Queue the command."""
@@ -1306,25 +1290,26 @@
     def store_marker_result(self, _):
         """Called when all the markers are realized."""
 
+    @defer.inlineCallbacks
     def run(self):
         """Do the deed."""
         self.running = True
-        if self.start_done:
-            self.log.debug('retrying')
-            d = defer.succeed(None)
+        try:
+            if self.start_done:
+                self.log.debug('retrying')
+            else:
+                self.log.debug('starting')
+                yield self._start()
+                self.start_done = True
+        except Exception, e:
+            yield self.end_errback(Failure(e))
         else:
-            self.log.debug('starting')
-            d = self._start()
-            d.addCallback(self.store_marker_result)
-        d.addCallbacks(self._ready_to_run, self.end_errback)
-        d.addBoth(self._done_running)
-        return d
-
-    def _done_running(self, x):
-        self.running = False
-        return x
-
-    def _ready_to_run(self, _):
+            yield self.markers_resolved_deferred
+            yield self._ready_to_run()
+        finally:
+            self.running = False
+
+    def _ready_to_run(self):
         self.log.debug('running')
         if self.running:
             d = self._run()
@@ -1426,25 +1411,22 @@
         self.name = name.decode("utf8")
         self.marker = marker
 
-    def _start(self):
-        """Do the specialized pre-run setup."""
-        return self.demark(self.share_id, self.parent_id)
+    def _get_possible_markers(self):
+        """Return the possible markers."""
+        return self.share_id, self.parent_id
 
     def store_marker_result(self, (share_id, parent_id)):
         """Called when all the markers are realized."""
         self.share_id = share_id
         self.parent_id = parent_id
-        self.start_done = True
 
     def _run(self):
         """Do the actual running."""
         maker = getattr(self.action_queue.client, self.client_method)
-        return maker(self.share_id,
-                     self.parent_id,
-                     self.name)
+        return maker(self.share_id, self.parent_id, self.name)
 
     def handle_success(self, request):
-        """It worked! Push the event, and update the uuid map."""
+        """It worked! Push the event."""
         # note that we're not getting the new name from the answer
         # message, if we would get it, we would have another Unicode
         # boundary with it
@@ -1452,20 +1434,13 @@
                  new_generation=request.new_generation,
                  volume_id=self.share_id)
         self.action_queue.event_queue.push(self.ok_event_name, **d)
-
-        # release the marker
-        if IMarker.providedBy(self.marker):
-            self.action_queue.uuid_map.set(self.marker, request.new_id)
         return request
 
     def handle_failure(self, failure):
-        """It didn't work! Push the event, and update the uuid map."""
+        """It didn't work! Push the event."""
         self.action_queue.event_queue.push(self.error_event_name,
                                            marker=self.marker,
-                                           error=failure.getErrorMessage())
-        if IMarker.providedBy(self.marker):
-            self.action_queue.uuid_map.err(self.marker,
-                                           failure)
+                                           failure=failure)
 
 
 class MakeFile(MakeThing):
@@ -1501,16 +1476,18 @@
         # here we use bytes for paths
         self.new_name = new_name.decode("utf8")
 
-    def _start(self):
-        """Do the specialized pre-run setup."""
-        return self.demark(self.share_id, self.node_id, self.new_parent_id)
+    def _get_possible_markers(self):
+        """Return the possible markers."""
+        return (self.share_id, self.node_id,
+                self.old_parent_id, self.new_parent_id)
 
-    def store_marker_result(self, (share_id, node_id, new_parent_id)):
+    def store_marker_result(self, (share_id, node_id,
+                                   old_parent_id, new_parent_id)):
         """Called when all the markers are realized."""
         self.share_id = share_id
         self.node_id = node_id
+        self.old_parent_id = old_parent_id
         self.new_parent_id = new_parent_id
-        self.start_done = True
 
     def _run(self):
         """Do the actual running."""
@@ -1550,16 +1527,15 @@
         self.node_id = node_id
         self.parent_id = parent_id
 
-    def _start(self):
-        """Do the specialized pre-run setup."""
-        return self.demark(self.share_id, self.node_id, self.parent_id)
+    def _get_possible_markers(self):
+        """Return the possible markers."""
+        return self.share_id, self.node_id, self.parent_id
 
     def store_marker_result(self, (share_id, node_id, parent_id)):
         """Called when all the markers are realized."""
         self.share_id = share_id
         self.node_id = node_id
         self.parent_id = parent_id
-        self.start_done = True
 
     def _run(self):
         """Do the actual running."""
@@ -1707,17 +1683,12 @@
             self.use_http = True
 
     def store_marker_result(self, (node_id,)):
-        """
-        Called when all the markers are realized.
-        """
+        """Called when all the markers are realized."""
         self.node_id = node_id
-        self.start_done = True
 
-    def _start(self):
-        """
-        Do the specialized pre-run setup
-        """
-        return self.demark(self.node_id)
+    def _get_possible_markers(self):
+        """Return the possible markers."""
+        return self.node_id,
 
     def _create_share_http(self, node_id, user, name, read_only, deferred):
         """Create a share using the HTTP Web API method."""
@@ -1879,6 +1850,7 @@
 
     __str__ = __repr__
 
+
 class GetDelta(ActionQueueCommand):
     """Gets a delta from a generation for a volume."""
 
@@ -2005,15 +1977,14 @@
         self.node_id = node_id
         self.is_public = is_public
 
-    def _start(self):
-        """See ActionQueueCommand."""
-        return self.demark(self.share_id, self.node_id)
+    def _get_possible_markers(self):
+        """Return the possible markers."""
+        return self.share_id, self.node_id
 
     def store_marker_result(self, (share_id, node_id)):
         """See ActionQueueCommand."""
         self.share_id = share_id
         self.node_id = node_id
-        self.start_done = True
 
     def _change_public_access_http(self):
         """Change public access using the HTTP Web API method."""
@@ -2166,9 +2137,9 @@
         self.download_req = None
         self.action_queue.cancel_download(self.share_id, self.node_id)
 
-    def _start(self):
-        """Do the specialized pre-run setup."""
-        return self.demark(self.node_id)
+    def _get_possible_markers(self):
+        """Return the possible markers."""
+        return self.share_id, self.node_id
 
     def cancel(self):
         """Cancel the download."""
@@ -2177,10 +2148,10 @@
             self.download_req.cancel()
         self.cleanup()
 
-    def store_marker_result(self, (node_id,)):
+    def store_marker_result(self, (share_id, node_id)):
         """Called when all the markers are realized."""
+        self.share_id = share_id
         self.node_id = node_id
-        self.start_done = True
 
     def _run(self):
         """Do the actual running."""
@@ -2365,6 +2336,10 @@
             self.log.debug('stopping the producer')
             self.upload_req.producer.stopProducing()
 
+    def _get_possible_markers(self):
+        """Return the possible markers."""
+        return self.share_id, self.node_id
+
     def _start(self):
         """Do the specialized pre-run setup."""
         d = defer.Deferred()
@@ -2373,19 +2348,14 @@
         self.action_queue.uploading[self.share_id, self.node_id] = uploading
 
         d = self.action_queue.zip_queue.zip(self)
-        d.addCallback(lambda _: self.demark(self.node_id))
         d.addBoth(lambda x: defer.fail(RequestCleanedUp('CANCELLED'))
                   if self.cancelled else x)
         return d
 
-    def store_marker_result(self, (node_id,)):
+    def store_marker_result(self, (share_id, node_id)):
         """Called when all the markers are realized."""
-        # update action_queue.uploading with the real node_id
-        uploading = self.action_queue.uploading.pop((self.share_id,
-                                                     self.node_id))
+        self.share_id = share_id
         self.node_id = node_id
-        self.action_queue.uploading[self.share_id, node_id] = uploading
-        self.start_done = True
 
     def _run(self):
         """Do the actual running."""

=== modified file 'ubuntuone/syncdaemon/event_queue.py'
--- ubuntuone/syncdaemon/event_queue.py	2010-08-17 14:23:31 +0000
+++ ubuntuone/syncdaemon/event_queue.py	2010-08-30 17:41:19 +0000
@@ -44,9 +44,9 @@
     'FS_INVALID_NAME': ('dirname', 'filename',),
 
     'AQ_FILE_NEW_OK': ('volume_id', 'marker', 'new_id', 'new_generation'),
-    'AQ_FILE_NEW_ERROR': ('marker', 'error'),
+    'AQ_FILE_NEW_ERROR': ('marker', 'failure'),
     'AQ_DIR_NEW_OK': ('volume_id', 'marker', 'new_id', 'new_generation'),
-    'AQ_DIR_NEW_ERROR': ('marker', 'error'),
+    'AQ_DIR_NEW_ERROR': ('marker', 'failure'),
     'AQ_MOVE_OK': ('share_id', 'node_id', 'new_generation'),
     'AQ_MOVE_ERROR': ('share_id', 'node_id',
                       'old_parent_id', 'new_parent_id', 'new_name', 'error'),

=== modified file 'ubuntuone/syncdaemon/sync.py'
--- ubuntuone/syncdaemon/sync.py	2010-08-10 14:08:40 +0000
+++ ubuntuone/syncdaemon/sync.py	2010-08-30 17:41:19 +0000
@@ -289,6 +289,7 @@
     debug = loglevel(logging.DEBUG)
     exception = loglevel(-1)
 
+
 class SyncStateMachineRunner(StateMachineRunner):
     """This is where all the state machine methods are."""
 
@@ -343,10 +344,10 @@
         params.update(self.build_hash_eq(hash))
         self.on_event(event, params, error, hash, *args)
 
-    def signal_event_with_error(self, event, error, *args):
+    def signal_event_with_error(self, event, failure, *args):
         """An event returned with error."""
-        params = self.build_error_eq(error)
-        self.on_event(event, params, error, *args)
+        params = self.build_error_eq(failure.getErrorMessage())
+        self.on_event(event, params, failure, *args)
 
     def build_error_eq(self, error):
         """Get the error state."""
@@ -541,7 +542,6 @@
 
     def new_local_file(self, event, parms, path):
         """a new local file was created"""
-        # XXX: lucio.torre: we should use markers here
         parent_path = os.path.dirname(path)
         parent = self.m.fs.get_by_path(parent_path)
         parent_id = parent.node_id or MDMarker(parent.mdid)
@@ -554,14 +554,21 @@
         marker = MDMarker(self.key.get_mdid())
         self.m.action_q.make_file(share_id, parent_id, name, marker)
 
-    def new_local_file_created(self, event, parms, new_id):
-        """we got the server answer for the file creation."""
+    def release_marker_ok(self, event, parms, new_id, marker):
+        """Release ok the received marker in AQ's DeferredMap."""
+        self.m.action_q.uuid_map.set(marker, new_id)
+
+    def release_marker_error(self, event, parms, failure, marker):
+        """Release with error the received marker in AQ's DeferredMap."""
+        self.m.action_q.uuid_map.err(marker, failure)
+
+    def new_local_file_created(self, event, parms, new_id, marker):
+        """We got the server answer for the file creation."""
+        self.m.action_q.uuid_map.set(marker, new_id)
         self.m.fs.set_node_id(self.key['path'], new_id)
 
-
     def new_local_dir(self, event, parms, path):
         """a new local dir was created"""
-        # XXX: lucio.torre: we should use markers here
         parent_path = os.path.dirname(path)
         parent = self.m.fs.get_by_path(parent_path)
         parent_id = parent.node_id or MDMarker(parent.mdid)
@@ -573,8 +580,9 @@
         self.m.action_q.make_dir(share_id, parent_id, name, marker)
         self.m.lr.scan_dir(mdid, path)
 
-    def new_local_dir_created(self, event, parms, new_id):
+    def new_local_dir_created(self, event, parms, new_id, marker):
         """Server answered that dir creation was ok."""
+        self.m.action_q.uuid_map.set(marker, new_id)
         self.m.fs.set_node_id(self.key['path'], new_id)
 
     def calculate_hash(self, event, params):
@@ -684,20 +692,12 @@
         self.key.remove_partial()
         self.delete_file(event, params)
 
-    def file_not_created_remove(self, event, params, error):
-        """kill it"""
+    def filedir_error_in_creation(self, event, params, failure, marker):
+        """Move actual content to conflict, and delete the metadata."""
+        self.m.action_q.uuid_map.err(marker, failure)
         self.key.move_to_conflict()
         self.key.delete_metadata()
 
-    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
-            self.key.delete_metadata()
-        else:
-            self.key.move_to_conflict()
-            self.key.delete_metadata()
-
     def delete_on_server(self, event, params, path):
         """local file was deleted."""
         self.m.action_q.unlink(self.key['share_id'],
@@ -935,29 +935,29 @@
         key = FSKey(self.m.fs, mdid=marker)
         log = FileLogger(self.logger, key)
         ssmr = SyncStateMachineRunner(self.fsm, self.m, key, log)
-        ssmr.on_event("AQ_FILE_NEW_OK", {}, new_id)
+        ssmr.on_event("AQ_FILE_NEW_OK", {}, new_id, marker)
         ssmr.update_generation(volume_id, new_id, new_generation)
 
-    def handle_AQ_FILE_NEW_ERROR(self, marker, error):
+    def handle_AQ_FILE_NEW_ERROR(self, marker, failure):
         """on AQ_FILE_NEW_ERROR"""
         key = FSKey(self.m.fs, mdid=marker)
         log = FileLogger(self.logger, key)
         ssmr = SyncStateMachineRunner(self.fsm, self.m, key, log)
-        ssmr.signal_event_with_error("AQ_FILE_NEW_ERROR", error)
+        ssmr.signal_event_with_error("AQ_FILE_NEW_ERROR", failure, marker)
 
-    def handle_AQ_DIR_NEW_ERROR(self, marker, error):
+    def handle_AQ_DIR_NEW_ERROR(self, marker, failure):
         """on AQ_DIR_NEW_ERROR"""
         key = FSKey(self.m.fs, mdid=marker)
         log = FileLogger(self.logger, key)
         ssmr = SyncStateMachineRunner(self.fsm, self.m, key, log)
-        ssmr.signal_event_with_error("AQ_DIR_NEW_ERROR", error)
+        ssmr.signal_event_with_error("AQ_DIR_NEW_ERROR", failure, marker)
 
     def handle_AQ_DIR_NEW_OK(self, volume_id, marker, new_id, new_generation):
         """On AQ_DIR_NEW_OK."""
         key = FSKey(self.m.fs, mdid=marker)
         log = FileLogger(self.logger, key)
         ssmr = SyncStateMachineRunner(self.fsm, self.m, key, log)
-        ssmr.on_event("AQ_DIR_NEW_OK", {}, new_id)
+        ssmr.on_event("AQ_DIR_NEW_OK", {}, new_id, marker)
         ssmr.update_generation(volume_id, new_id, new_generation)
 
     def handle_FS_FILE_CLOSE_WRITE(self, path):

=== modified file 'ubuntuone/syncdaemon/u1fsfsm.ods'
Binary files ubuntuone/syncdaemon/u1fsfsm.ods	2010-07-16 21:36:52 +0000 and ubuntuone/syncdaemon/u1fsfsm.ods	2010-08-30 17:41:19 +0000 differ
=== modified file 'ubuntuone/syncdaemon/u1fsfsm.py'
--- ubuntuone/syncdaemon/u1fsfsm.py	2010-07-19 17:31:18 +0000
+++ ubuntuone/syncdaemon/u1fsfsm.py	2010-08-30 17:41:19 +0000
@@ -210,7 +210,7 @@
                                                  u'has_metadata': u'=',
                                                  u'is_directory': u'='}}],
             u'AQ_DIR_NEW_ERROR': [{'ACTION': u'pass',
-                                   'ACTION_FUNC': u'nothing',
+                                   'ACTION_FUNC': u'release_marker_error',
                                    'COMMENTS': u'',
                                    'PARAMETERS': {u'hash_eq_local_hash': u'NA',
                                                   u'hash_eq_server_hash': u'NA',
@@ -223,24 +223,11 @@
                                                  u'has_metadata': u'=',
                                                  u'is_directory': u'='}},
                                   {'ACTION': u'CONFLICT',
-                                   'ACTION_FUNC': u'file_not_created_remove',
-                                   'COMMENTS': u'',
-                                   'PARAMETERS': {u'hash_eq_local_hash': u'NA',
-                                                  u'hash_eq_server_hash': u'NA',
-                                                  u'not_authorized': u'T',
-                                                  u'not_available': u'*'},
-                                   'STATE': {u'changed': u'*',
-                                             u'has_metadata': u'*',
-                                             u'is_directory': u'T'},
-                                   'STATE_OUT': {u'changed': u'NA',
-                                                 u'has_metadata': u'F',
-                                                 u'is_directory': u'NA'}},
-                                  {'ACTION': u'if the error is because the dir already exists in the server, we need to merge them',
-                                   'ACTION_FUNC': u'dir_not_created_in_server',
-                                   'COMMENTS': u'',
-                                   'PARAMETERS': {u'hash_eq_local_hash': u'NA',
-                                                  u'hash_eq_server_hash': u'NA',
-                                                  u'not_authorized': u'F',
+                                   'ACTION_FUNC': u'filedir_error_in_creation',
+                                   'COMMENTS': u'',
+                                   'PARAMETERS': {u'hash_eq_local_hash': u'NA',
+                                                  u'hash_eq_server_hash': u'NA',
+                                                  u'not_authorized': u'*',
                                                   u'not_available': u'*'},
                                    'STATE': {u'changed': u'*',
                                              u'has_metadata': u'*',
@@ -249,7 +236,7 @@
                                                  u'has_metadata': u'F',
                                                  u'is_directory': u'NA'}},
                                   {'ACTION': u'pass',
-                                   'ACTION_FUNC': u'nothing',
+                                   'ACTION_FUNC': u'release_marker_error',
                                    'COMMENTS': u'',
                                    'PARAMETERS': {u'hash_eq_local_hash': u'NA',
                                                   u'hash_eq_server_hash': u'NA',
@@ -260,48 +247,22 @@
                                              u'is_directory': u'F'},
                                    'STATE_OUT': {u'changed': u'=',
                                                  u'has_metadata': u'=',
-                                                 u'is_directory': u'='}},
-                                  {'ACTION': u'NA',
-                                   'ACTION_FUNC': u'',
-                                   'COMMENTS': u'',
-                                   'PARAMETERS': {u'hash_eq_local_hash': u'NA',
-                                                  u'hash_eq_server_hash': u'NA',
-                                                  u'not_authorized': u'NA',
-                                                  u'not_available': u'*'},
-                                   'STATE': {u'changed': u'*',
-                                             u'has_metadata': u'*',
-                                             u'is_directory': u'T'},
-                                   'STATE_OUT': {u'changed': u'=',
-                                                 u'has_metadata': u'=',
                                                  u'is_directory': u'='}}],
-            u'AQ_DIR_NEW_OK': [{'ACTION': u'',
-                                'ACTION_FUNC': u'nothing',
-                                'COMMENTS': u'',
-                                'PARAMETERS': {u'hash_eq_local_hash': u'NA',
-                                               u'hash_eq_server_hash': u'NA',
-                                               u'not_authorized': u'NA',
-                                               u'not_available': u'NA'},
-                                'STATE': {u'changed': u'NONE',
-                                          u'has_metadata': u'T',
-                                          u'is_directory': u'F'},
-                                'STATE_OUT': {u'changed': u'=',
-                                              u'has_metadata': u'=',
-                                              u'is_directory': u'='}},
-                               {'ACTION': u'pass',
-                                'ACTION_FUNC': u'nothing',
-                                'COMMENTS': u'we got IN_FILE_CHANGED and HQ_HASH_NEW between IN_FILE_NEW and AQ_FILE_NEW_OK',
-                                'PARAMETERS': {u'hash_eq_local_hash': u'NA',
-                                               u'hash_eq_server_hash': u'NA',
-                                               u'not_authorized': u'NA',
-                                               u'not_available': u'NA'},
-                                'STATE': {u'changed': u'LOCAL',
-                                          u'has_metadata': u'T',
-                                          u'is_directory': u'F'},
-                                'STATE_OUT': {u'changed': u'=',
-                                              u'has_metadata': u'=',
-                                              u'is_directory': u'='}},
-                               {'ACTION': u'',
-                                'ACTION_FUNC': u'nothing',
+            u'AQ_DIR_NEW_OK': [{'ACTION': u'aq.uuid_map.set(marker, new_id)',
+                                'ACTION_FUNC': u'release_marker_ok',
+                                'COMMENTS': u"it's a file now",
+                                'PARAMETERS': {u'hash_eq_local_hash': u'NA',
+                                               u'hash_eq_server_hash': u'NA',
+                                               u'not_authorized': u'NA',
+                                               u'not_available': u'NA'},
+                                'STATE': {u'changed': u'*',
+                                          u'has_metadata': u'T',
+                                          u'is_directory': u'F'},
+                                'STATE_OUT': {u'changed': u'=',
+                                              u'has_metadata': u'=',
+                                              u'is_directory': u'='}},
+                               {'ACTION': u'aq.uuid_map.set(marker, new_id)',
+                                'ACTION_FUNC': u'release_marker_ok',
                                 'COMMENTS': u'the dir was now gone',
                                 'PARAMETERS': {u'hash_eq_local_hash': u'NA',
                                                u'hash_eq_server_hash': u'NA',
@@ -322,7 +283,7 @@
                                                u'not_available': u'NA'},
                                 'STATE': {u'changed': u'SERVER',
                                           u'has_metadata': u'T',
-                                          u'is_directory': u'*'},
+                                          u'is_directory': u'T'},
                                 'STATE_OUT': {u'changed': u'NONE',
                                               u'has_metadata': u'T',
                                               u'is_directory': u'='}},
@@ -977,7 +938,7 @@
                                                 u'has_metadata': u'T',
                                                 u'is_directory': u'='}}],
             u'AQ_FILE_NEW_ERROR': [{'ACTION': u'pass',
-                                    'ACTION_FUNC': u'nothing',
+                                    'ACTION_FUNC': u'release_marker_error',
                                     'COMMENTS': u'',
                                     'PARAMETERS': {u'hash_eq_local_hash': u'NA',
                                                    u'hash_eq_server_hash': u'NA',
@@ -990,24 +951,11 @@
                                                   u'has_metadata': u'=',
                                                   u'is_directory': u'='}},
                                    {'ACTION': u'CONFLICT',
-                                    'ACTION_FUNC': u'file_not_created_remove',
-                                    'COMMENTS': u'',
-                                    'PARAMETERS': {u'hash_eq_local_hash': u'NA',
-                                                   u'hash_eq_server_hash': u'NA',
-                                                   u'not_authorized': u'T',
-                                                   u'not_available': u'*'},
-                                    'STATE': {u'changed': u'*',
-                                              u'has_metadata': u'*',
-                                              u'is_directory': u'F'},
-                                    'STATE_OUT': {u'changed': u'NA',
-                                                  u'has_metadata': u'F',
-                                                  u'is_directory': u'NA'}},
-                                   {'ACTION': u'CONFLICT',
-                                    'ACTION_FUNC': u'file_not_created_remove',
-                                    'COMMENTS': u'',
-                                    'PARAMETERS': {u'hash_eq_local_hash': u'NA',
-                                                   u'hash_eq_server_hash': u'NA',
-                                                   u'not_authorized': u'F',
+                                    'ACTION_FUNC': u'filedir_error_in_creation',
+                                    'COMMENTS': u'',
+                                    'PARAMETERS': {u'hash_eq_local_hash': u'NA',
+                                                   u'hash_eq_server_hash': u'NA',
+                                                   u'not_authorized': u'*',
                                                    u'not_available': u'*'},
                                     'STATE': {u'changed': u'*',
                                               u'has_metadata': u'*',
@@ -1016,7 +964,7 @@
                                                   u'has_metadata': u'F',
                                                   u'is_directory': u'NA'}},
                                    {'ACTION': u'pass',
-                                    'ACTION_FUNC': u'nothing',
+                                    'ACTION_FUNC': u'release_marker_error',
                                     'COMMENTS': u'',
                                     'PARAMETERS': {u'hash_eq_local_hash': u'NA',
                                                    u'hash_eq_server_hash': u'NA',
@@ -1027,19 +975,6 @@
                                               u'is_directory': u'T'},
                                     'STATE_OUT': {u'changed': u'=',
                                                   u'has_metadata': u'=',
-                                                  u'is_directory': u'='}},
-                                   {'ACTION': u'NA',
-                                    'ACTION_FUNC': u'',
-                                    'COMMENTS': u'',
-                                    'PARAMETERS': {u'hash_eq_local_hash': u'NA',
-                                                   u'hash_eq_server_hash': u'NA',
-                                                   u'not_authorized': u'NA',
-                                                   u'not_available': u'*'},
-                                    'STATE': {u'changed': u'*',
-                                              u'has_metadata': u'*',
-                                              u'is_directory': u'F'},
-                                    'STATE_OUT': {u'changed': u'=',
-                                                  u'has_metadata': u'=',
                                                   u'is_directory': u'='}}],
             u'AQ_FILE_NEW_OK': [{'ACTION': u'md.set(mdid, server_uuid=server_uuid)',
                                  'ACTION_FUNC': u'new_local_file_created',
@@ -1067,8 +1002,8 @@
                                  'STATE_OUT': {u'changed': u'=',
                                                u'has_metadata': u'=',
                                                u'is_directory': u'='}},
-                                {'ACTION': u'pass',
-                                 'ACTION_FUNC': u'nothing',
+                                {'ACTION': u'aq.uuid_map.set(marker, new_id)',
+                                 'ACTION_FUNC': u'release_marker_ok',
                                  'COMMENTS': u'file deleted locally',
                                  'PARAMETERS': {u'hash_eq_local_hash': u'NA',
                                                 u'hash_eq_server_hash': u'NA',
@@ -1089,18 +1024,18 @@
                                                 u'not_available': u'NA'},
                                  'STATE': {u'changed': u'SERVER',
                                            u'has_metadata': u'T',
-                                           u'is_directory': u'*'},
+                                           u'is_directory': u'F'},
                                  'STATE_OUT': {u'changed': u'NONE',
                                                u'has_metadata': u'T',
                                                u'is_directory': u'='}},
-                                {'ACTION': u'pass',
-                                 'ACTION_FUNC': u'nothing',
-                                 'COMMENTS': u'',
+                                {'ACTION': u'aq.uuid_map.set(marker, new_id)',
+                                 'ACTION_FUNC': u'release_marker_ok',
+                                 'COMMENTS': u"it's a directory now",
                                  'PARAMETERS': {u'hash_eq_local_hash': u'NA',
                                                 u'hash_eq_server_hash': u'NA',
                                                 u'not_authorized': u'NA',
                                                 u'not_available': u'NA'},
-                                 'STATE': {u'changed': u'NONE',
+                                 'STATE': {u'changed': u'*',
                                            u'has_metadata': u'T',
                                            u'is_directory': u'T'},
                                  'STATE_OUT': {u'changed': u'=',