src/allmydata/frontends/magic_folder.py

   1
   2 import sys, os
   3 import os.path
   4 from collections import deque
   5 import time
   6
   7 from twisted.internet import defer, reactor, task
   8 from twisted.python.failure import Failure
   9 from twisted.python import runtime
  10 from twisted.application import service
  11
  12 from allmydata.util import fileutil
  13 from allmydata.interfaces import IDirectoryNode
  14 from allmydata.util import log
  15 from allmydata.util.fileutil import precondition_abspath, get_pathinfo, ConflictError
  16 from allmydata.util.assertutil import precondition, _assert
  17 from allmydata.util.deferredutil import HookMixin
  18 from allmydata.util.encodingutil import listdir_filepath, to_filepath, \
  19      extend_filepath, unicode_from_filepath, unicode_segments_from, \
  20      quote_filepath, quote_local_unicode_path, quote_output, FilenameEncodingError
  21 from allmydata.immutable.upload import FileName, Data
  22 from allmydata import magicfolderdb, magicpath
  23
  24
  25 IN_EXCL_UNLINK = 0x04000000L
  26
  27 def get_inotify_module():
  28     try:
  29         if sys.platform == "win32":
  30             from allmydata.windows import inotify
  31         elif runtime.platform.supportsINotify():
  32             from twisted.internet import inotify
  33         else:
  34             raise NotImplementedError("filesystem notification needed for Magic Folder is not supported.\n"
  35                                       "This currently requires Linux or Windows.")
  36         return inotify
  37     except (ImportError, AttributeError) as e:
  38         log.msg(e)
  39         if sys.platform == "win32":
  40             raise NotImplementedError("filesystem notification needed for Magic Folder is not supported.\n"
  41                                       "Windows support requires at least Vista, and has only been tested on Windows 7.")
  42         raise
  43
  44
  45 class MagicFolder(service.MultiService):
  46     name = 'magic-folder'
  47
  48     def __init__(self, client, upload_dircap, collective_dircap, local_path_u, dbfile,
  49                  pending_delay=1.0, clock=reactor):
  50         precondition_abspath(local_path_u)
  51
  52         service.MultiService.__init__(self)
  53
  54         db = magicfolderdb.get_magicfolderdb(dbfile, create_version=(magicfolderdb.SCHEMA_v1, 1))
  55         if db is None:
  56             return Failure(Exception('ERROR: Unable to load magic folder db.'))
  57
  58         # for tests
  59         self._client = client
  60         self._db = db
  61
  62         self.is_ready = False
  63
  64         self.uploader = Uploader(client, local_path_u, db, upload_dircap, pending_delay, clock)
  65         self.downloader = Downloader(client, local_path_u, db, collective_dircap, clock)
  66
  67     def startService(self):
  68         # TODO: why is this being called more than once?
  69         if self.running:
  70             return defer.succeed(None)
  71         print "%r.startService" % (self,)
  72         service.MultiService.startService(self)
  73         return self.uploader.start_monitoring()
  74
  75     def ready(self):
  76         """ready is used to signal us to start
  77         processing the upload and download items...
  78         """
  79         self.is_ready = True
  80         d = self.uploader.start_scanning()
  81         d2 = self.downloader.start_scanning()
  82         d.addCallback(lambda ign: d2)
  83         return d
  84
  85     def finish(self):
  86         print "finish"
  87         d = self.uploader.stop()
  88         d2 = self.downloader.stop()
  89         d.addCallback(lambda ign: d2)
  90         return d
  91
  92     def remove_service(self):
  93         return service.MultiService.disownServiceParent(self)
  94
  95
  96 class QueueMixin(HookMixin):
  97     def __init__(self, client, local_path_u, db, name, clock):
  98         self._client = client
  99         self._local_path_u = local_path_u
 100         self._local_filepath = to_filepath(local_path_u)
 101         self._db = db
 102         self._name = name
 103         self._clock = clock
 104         self._hooks = {'processed': None, 'started': None}
 105         self.started_d = self.set_hook('started')
 106
 107         if not self._local_filepath.exists():
 108             raise AssertionError("The '[magic_folder] local.directory' parameter was %s "
 109                                  "but there is no directory at that location."
 110                                  % quote_local_unicode_path(self._local_path_u))
 111         if not self._local_filepath.isdir():
 112             raise AssertionError("The '[magic_folder] local.directory' parameter was %s "
 113                                  "but the thing at that location is not a directory."
 114                                  % quote_local_unicode_path(self._local_path_u))
 115
 116         self._deque = deque()
 117         self._lazy_tail = defer.succeed(None)
 118         self._pending = set()
 119         self._stopped = False
 120         self._turn_delay = 0
 121
 122     def _get_filepath(self, relpath_u):
 123         return extend_filepath(self._local_filepath, relpath_u.split(u"/"))
 124
 125     def _get_relpath(self, filepath):
 126         self._log("_get_relpath(%r)" % (filepath,))
 127         segments = unicode_segments_from(filepath, self._local_filepath)
 128         self._log("segments = %r" % (segments,))
 129         return u"/".join(segments)
 130
 131     def _count(self, counter_name, delta=1):
 132         ctr = 'magic_folder.%s.%s' % (self._name, counter_name)
 133         self._log("%s += %r" % (counter_name, delta))
 134         self._client.stats_provider.count(ctr, delta)
 135
 136     def _logcb(self, res, msg):
 137         self._log("%s: %r" % (msg, res))
 138         return res
 139
 140     def _log(self, msg):
 141         s = "Magic Folder %s %s: %s" % (quote_output(self._client.nickname), self._name, msg)
 142         self._client.log(s)
 143         print s
 144         #open("events", "ab+").write(msg)
 145
 146     def _append_to_deque(self, relpath_u):
 147         self._log("_append_to_deque(%r)" % (relpath_u,))
 148         if relpath_u in self._pending or magicpath.should_ignore_file(relpath_u):
 149             return
 150         self._deque.append(relpath_u)
 151         self._pending.add(relpath_u)
 152         self._count('objects_queued')
 153         if self.is_ready:
 154             self._clock.callLater(0, self._turn_deque)
 155
 156     def _turn_deque(self):
 157         self._log("_turn_deque")
 158         if self._stopped:
 159             self._log("stopped")
 160             return
 161         try:
 162             item = self._deque.pop()
 163             self._log("popped %r" % (item,))
 164             self._count('objects_queued', -1)
 165         except IndexError:
 166             self._log("deque is now empty")
 167             self._lazy_tail.addCallback(lambda ign: self._when_queue_is_empty())
 168         else:
 169             self._lazy_tail.addCallback(lambda ign: self._process(item))
 170             self._lazy_tail.addBoth(self._call_hook, 'processed')
 171             self._lazy_tail.addErrback(log.err)
 172             self._lazy_tail.addCallback(lambda ign: task.deferLater(self._clock, self._turn_delay, self._turn_deque))
 173
 174
 175 class Uploader(QueueMixin):
 176     def __init__(self, client, local_path_u, db, upload_dircap, pending_delay, clock):
 177         QueueMixin.__init__(self, client, local_path_u, db, 'uploader', clock)
 178
 179         self.is_ready = False
 180
 181         # TODO: allow a path rather than a cap URI.
 182         self._upload_dirnode = self._client.create_node_from_uri(upload_dircap)
 183         if not IDirectoryNode.providedBy(self._upload_dirnode):
 184             raise AssertionError("The URI in '%s' does not refer to a directory."
 185                                  % os.path.join('private', 'magic_folder_dircap'))
 186         if self._upload_dirnode.is_unknown() or self._upload_dirnode.is_readonly():
 187             raise AssertionError("The URI in '%s' is not a writecap to a directory."
 188                                  % os.path.join('private', 'magic_folder_dircap'))
 189
 190         self._inotify = get_inotify_module()
 191         self._notifier = self._inotify.INotify()
 192
 193         if hasattr(self._notifier, 'set_pending_delay'):
 194             self._notifier.set_pending_delay(pending_delay)
 195
 196         # TODO: what about IN_MOVE_SELF, IN_MOVED_FROM, or IN_UNMOUNT?
 197         #
 198         self.mask = ( self._inotify.IN_CREATE
 199                     | self._inotify.IN_CLOSE_WRITE
 200                     | self._inotify.IN_MOVED_TO
 201                     | self._inotify.IN_MOVED_FROM
 202                     | self._inotify.IN_DELETE
 203                     | self._inotify.IN_ONLYDIR
 204                     | IN_EXCL_UNLINK
 205                     )
 206         self._notifier.watch(self._local_filepath, mask=self.mask, callbacks=[self._notify],
 207                              recursive=True)
 208
 209     def start_monitoring(self):
 210         self._log("start_monitoring")
 211         d = defer.succeed(None)
 212         d.addCallback(lambda ign: self._notifier.startReading())
 213         d.addCallback(lambda ign: self._count('dirs_monitored'))
 214         d.addBoth(self._call_hook, 'started')
 215         return d
 216
 217     def stop(self):
 218         self._log("stop")
 219         self._notifier.stopReading()
 220         self._count('dirs_monitored', -1)
 221         if hasattr(self._notifier, 'wait_until_stopped'):
 222             d = self._notifier.wait_until_stopped()
 223         else:
 224             d = defer.succeed(None)
 225         d.addCallback(lambda ign: self._lazy_tail)
 226         return d
 227
 228     def start_scanning(self):
 229         self._log("start_scanning")
 230         self.is_ready = True
 231         self._pending = self._db.get_all_relpaths()
 232         self._log("all_files %r" % (self._pending))
 233         d = self._scan(u"")
 234         def _add_pending(ign):
 235             # This adds all of the files that were in the db but not already processed
 236             # (normally because they have been deleted on disk).
 237             self._log("adding %r" % (self._pending))
 238             self._deque.extend(self._pending)
 239         d.addCallback(_add_pending)
 240         d.addCallback(lambda ign: self._turn_deque())
 241         return d
 242
 243     def _scan(self, reldir_u):
 244         self._log("scan %r" % (reldir_u,))
 245         fp = self._get_filepath(reldir_u)
 246         try:
 247             children = listdir_filepath(fp)
 248         except EnvironmentError:
 249             raise Exception("WARNING: magic folder: permission denied on directory %s"
 250                             % quote_filepath(fp))
 251         except FilenameEncodingError:
 252             raise Exception("WARNING: magic folder: could not list directory %s due to a filename encoding error"
 253                             % quote_filepath(fp))
 254
 255         d = defer.succeed(None)
 256         for child in children:
 257             _assert(isinstance(child, unicode), child=child)
 258             d.addCallback(lambda ign, child=child:
 259                           ("%s/%s" % (reldir_u, child) if reldir_u else child))
 260             def _add_pending(relpath_u):
 261                 if magicpath.should_ignore_file(relpath_u):
 262                     return None
 263
 264                 self._pending.add(relpath_u)
 265                 return relpath_u
 266             d.addCallback(_add_pending)
 267             # This call to _process doesn't go through the deque, and probably should.
 268             d.addCallback(self._process)
 269             d.addBoth(self._call_hook, 'processed')
 270             d.addErrback(log.err)
 271
 272         return d
 273
 274     def _notify(self, opaque, path, events_mask):
 275         self._log("inotify event %r, %r, %r\n" % (opaque, path, ', '.join(self._inotify.humanReadableMask(events_mask))))
 276
 277         # We filter out IN_CREATE events not associated with a directory.
 278         # Acting on IN_CREATE for files could cause us to read and upload
 279         # a possibly-incomplete file before the application has closed it.
 280         # There should always be an IN_CLOSE_WRITE after an IN_CREATE, I think.
 281         # It isn't possible to avoid watching for IN_CREATE at all, because
 282         # it is the only event notified for a directory creation.
 283
 284         if ((events_mask & self._inotify.IN_CREATE) != 0 and
 285             (events_mask & self._inotify.IN_ISDIR) == 0):
 286             self._log("ignoring inotify event for creation of file %r\n" % (path,))
 287             return
 288
 289         relpath_u = self._get_relpath(path)
 290         self._append_to_deque(relpath_u)
 291
 292     def _when_queue_is_empty(self):
 293         return defer.succeed(None)
 294
 295     def _process(self, relpath_u):
 296         self._log("_process(%r)" % (relpath_u,))
 297         if relpath_u is None:
 298             return
 299         precondition(isinstance(relpath_u, unicode), relpath_u)
 300
 301         d = defer.succeed(None)
 302
 303         def _maybe_upload(val, now=None):
 304             if now is None:
 305                 now = time.time()
 306             fp = self._get_filepath(relpath_u)
 307             pathinfo = get_pathinfo(unicode_from_filepath(fp))
 308
 309             self._log("pending = %r, about to remove %r" % (self._pending, relpath_u))
 310             self._pending.remove(relpath_u)
 311             encoded_path_u = magicpath.path2magic(relpath_u)
 312
 313             if not pathinfo.exists:
 314                 # FIXME merge this with the 'isfile' case.
 315                 self._log("notified object %s disappeared (this is normal)" % quote_filepath(fp))
 316                 self._count('objects_disappeared')
 317                 if not self._db.check_file_db_exists(relpath_u):
 318                     return None
 319
 320                 last_downloaded_timestamp = now
 321                 last_downloaded_uri = self._db.get_last_downloaded_uri(relpath_u)
 322
 323                 current_version = self._db.get_local_file_version(relpath_u)
 324                 if current_version is None:
 325                     new_version = 0
 326                 elif self._db.is_new_file(pathinfo, relpath_u):
 327                     new_version = current_version + 1
 328                 else:
 329                     self._log("Not uploading %r" % (relpath_u,))
 330                     self._count('objects_not_uploaded')
 331                     return
 332
 333                 metadata = { 'version': new_version,
 334                              'deleted': True,
 335                              'last_downloaded_timestamp': last_downloaded_timestamp }
 336                 if last_downloaded_uri is not None:
 337                     metadata['last_downloaded_uri'] = last_downloaded_uri
 338
 339                 empty_uploadable = Data("", self._client.convergence)
 340                 d2 = self._upload_dirnode.add_file(encoded_path_u, empty_uploadable,
 341                                                    metadata=metadata, overwrite=True)
 342
 343                 def _add_db_entry(filenode):
 344                     filecap = filenode.get_uri()
 345                     self._db.did_upload_version(relpath_u, new_version, filecap,
 346                                                 last_downloaded_uri, last_downloaded_timestamp, pathinfo)
 347                     self._count('files_uploaded')
 348                 d2.addCallback(_add_db_entry)
 349                 return d2
 350             elif pathinfo.islink:
 351                 self.warn("WARNING: cannot upload symlink %s" % quote_filepath(fp))
 352                 return None
 353             elif pathinfo.isdir:
 354                 self._notifier.watch(fp, mask=self.mask, callbacks=[self._notify], recursive=True)
 355                 uploadable = Data("", self._client.convergence)
 356                 encoded_path_u += magicpath.path2magic(u"/")
 357                 upload_d = self._upload_dirnode.add_file(encoded_path_u, uploadable, metadata={"version":0}, overwrite=True)
 358                 def _succeeded(ign):
 359                     self._log("created subdirectory %r" % (relpath_u,))
 360                     self._count('directories_created')
 361                 def _failed(f):
 362                     self._log("failed to create subdirectory %r" % (relpath_u,))
 363                     return f
 364                 upload_d.addCallbacks(_succeeded, _failed)
 365                 upload_d.addCallback(lambda ign: self._scan(relpath_u))
 366                 return upload_d
 367             elif pathinfo.isfile:
 368                 last_downloaded_uri = self._db.get_last_downloaded_uri(relpath_u)
 369                 last_downloaded_timestamp = now
 370
 371                 current_version = self._db.get_local_file_version(relpath_u)
 372                 if current_version is None:
 373                     new_version = 0
 374                 elif self._db.is_new_file(pathinfo, relpath_u):
 375                     new_version = current_version + 1
 376                 else:
 377                     self._log("Not uploading %r" % (relpath_u,))
 378                     self._count('objects_not_uploaded')
 379                     return None
 380
 381                 metadata = { 'version': new_version,
 382                              'last_downloaded_timestamp': last_downloaded_timestamp }
 383                 if last_downloaded_uri is not None:
 384                     metadata['last_downloaded_uri'] = last_downloaded_uri
 385
 386                 uploadable = FileName(unicode_from_filepath(fp), self._client.convergence)
 387                 d2 = self._upload_dirnode.add_file(encoded_path_u, uploadable,
 388                                                    metadata=metadata, overwrite=True)
 389
 390                 def _add_db_entry(filenode):
 391                     filecap = filenode.get_uri()
 392                     last_downloaded_uri = metadata.get('last_downloaded_uri', None)
 393                     self._db.did_upload_version(relpath_u, new_version, filecap,
 394                                                 last_downloaded_uri, last_downloaded_timestamp, pathinfo)
 395                     self._count('files_uploaded')
 396                 d2.addCallback(_add_db_entry)
 397                 return d2
 398             else:
 399                 self.warn("WARNING: cannot process special file %s" % quote_filepath(fp))
 400                 return None
 401
 402         d.addCallback(_maybe_upload)
 403
 404         def _succeeded(res):
 405             self._count('objects_succeeded')
 406             return res
 407         def _failed(f):
 408             self._count('objects_failed')
 409             self._log("%s while processing %r" % (f, relpath_u))
 410             return f
 411         d.addCallbacks(_succeeded, _failed)
 412         return d
 413
 414     def _get_metadata(self, encoded_path_u):
 415         try:
 416             d = self._upload_dirnode.get_metadata_for(encoded_path_u)
 417         except KeyError:
 418             return Failure()
 419         return d
 420
 421     def _get_filenode(self, encoded_path_u):
 422         try:
 423             d = self._upload_dirnode.get(encoded_path_u)
 424         except KeyError:
 425             return Failure()
 426         return d
 427
 428
 429 class WriteFileMixin(object):
 430     FUDGE_SECONDS = 10.0
 431
 432     def _get_conflicted_filename(self, abspath_u):
 433         return abspath_u + u".conflict"
 434
 435     def _write_downloaded_file(self, abspath_u, file_contents, is_conflict=False, now=None):
 436         self._log("_write_downloaded_file(%r, <%d bytes>, is_conflict=%r, now=%r)"
 437                   % (abspath_u, len(file_contents), is_conflict, now))
 438
 439         # 1. Write a temporary file, say .foo.tmp.
 440         # 2. is_conflict determines whether this is an overwrite or a conflict.
 441         # 3. Set the mtime of the replacement file to be T seconds before the
 442         #    current local time.
 443         # 4. Perform a file replacement with backup filename foo.backup,
 444         #    replaced file foo, and replacement file .foo.tmp. If any step of
 445         #    this operation fails, reclassify as a conflict and stop.
 446         #
 447         # Returns the path of the destination file.
 448
 449         precondition_abspath(abspath_u)
 450         replacement_path_u = abspath_u + u".tmp"  # FIXME more unique
 451         backup_path_u = abspath_u + u".backup"
 452         if now is None:
 453             now = time.time()
 454
 455         # ensure parent directory exists
 456         head, tail = os.path.split(abspath_u)
 457         mode = 0777 # XXX
 458         fileutil.make_dirs(head, mode)
 459
 460         fileutil.write(replacement_path_u, file_contents)
 461         os.utime(replacement_path_u, (now, now - self.FUDGE_SECONDS))
 462         if is_conflict:
 463             print "0x00 ------------ <><> is conflict; calling _rename_conflicted_file... %r %r" % (abspath_u, replacement_path_u)
 464             return self._rename_conflicted_file(abspath_u, replacement_path_u)
 465         else:
 466             try:
 467                 fileutil.replace_file(abspath_u, replacement_path_u, backup_path_u)
 468                 return abspath_u
 469             except fileutil.ConflictError:
 470                 return self._rename_conflicted_file(abspath_u, replacement_path_u)
 471
 472     def _rename_conflicted_file(self, abspath_u, replacement_path_u):
 473         self._log("_rename_conflicted_file(%r, %r)" % (abspath_u, replacement_path_u))
 474
 475         conflict_path_u = self._get_conflicted_filename(abspath_u)
 476         print "XXX rename %r %r" % (replacement_path_u, conflict_path_u)
 477         if os.path.isfile(replacement_path_u):
 478             print "%r exists" % (replacement_path_u,)
 479         if os.path.isfile(conflict_path_u):
 480             print "%r exists" % (conflict_path_u,)
 481
 482         fileutil.rename_no_overwrite(replacement_path_u, conflict_path_u)
 483         return conflict_path_u
 484
 485     def _rename_deleted_file(self, abspath_u):
 486         self._log('renaming deleted file to backup: %s' % (abspath_u,))
 487         try:
 488             fileutil.rename_no_overwrite(abspath_u, abspath_u + u'.backup')
 489         except IOError:
 490             # XXX is this the correct error?
 491             self._log("Already gone: '%s'" % (abspath_u,))
 492         return abspath_u
 493
 494
 495 class Downloader(QueueMixin, WriteFileMixin):
 496     REMOTE_SCAN_INTERVAL = 3  # facilitates tests
 497
 498     def __init__(self, client, local_path_u, db, collective_dircap, clock):
 499         QueueMixin.__init__(self, client, local_path_u, db, 'downloader', clock)
 500
 501         # TODO: allow a path rather than a cap URI.
 502         self._collective_dirnode = self._client.create_node_from_uri(collective_dircap)
 503
 504         if not IDirectoryNode.providedBy(self._collective_dirnode):
 505             raise AssertionError("The URI in '%s' does not refer to a directory."
 506                                  % os.path.join('private', 'collective_dircap'))
 507         if self._collective_dirnode.is_unknown() or not self._collective_dirnode.is_readonly():
 508             raise AssertionError("The URI in '%s' is not a readonly cap to a directory."
 509                                  % os.path.join('private', 'collective_dircap'))
 510
 511         self._turn_delay = self.REMOTE_SCAN_INTERVAL
 512         self._download_scan_batch = {} # path -> [(filenode, metadata)]
 513
 514     def start_scanning(self):
 515         self._log("start_scanning")
 516         files = self._db.get_all_relpaths()
 517         self._log("all files %s" % files)
 518
 519         d = self._scan_remote_collective()
 520         self._turn_deque()
 521         return d
 522
 523     def stop(self):
 524         self._stopped = True
 525         d = defer.succeed(None)
 526         d.addCallback(lambda ign: self._lazy_tail)
 527         return d
 528
 529     def _should_download(self, relpath_u, remote_version):
 530         """
 531         _should_download returns a bool indicating whether or not a remote object should be downloaded.
 532         We check the remote metadata version against our magic-folder db version number;
 533         latest version wins.
 534         """
 535         self._log("_should_download(%r, %r)" % (relpath_u, remote_version))
 536         if magicpath.should_ignore_file(relpath_u):
 537             self._log("nope")
 538             return False
 539         self._log("yep")
 540         v = self._db.get_local_file_version(relpath_u)
 541         self._log("v = %r" % (v,))
 542         return (v is None or v < remote_version)
 543
 544     def _get_local_latest(self, relpath_u):
 545         """
 546         _get_local_latest takes a unicode path string checks to see if this file object
 547         exists in our magic-folder db; if not then return None
 548         else check for an entry in our magic-folder db and return the version number.
 549         """
 550         if not self._get_filepath(relpath_u).exists():
 551             return None
 552         return self._db.get_local_file_version(relpath_u)
 553
 554     def _get_collective_latest_file(self, filename):
 555         """
 556         _get_collective_latest_file takes a file path pointing to a file managed by
 557         magic-folder and returns a deferred that fires with the two tuple containing a
 558         file node and metadata for the latest version of the file located in the
 559         magic-folder collective directory.
 560         """
 561         collective_dirmap_d = self._collective_dirnode.list()
 562         def scan_collective(result):
 563             list_of_deferreds = []
 564             for dir_name in result.keys():
 565                 # XXX make sure it's a directory
 566                 d = defer.succeed(None)
 567                 d.addCallback(lambda x, dir_name=dir_name: result[dir_name][0].get_child_and_metadata(filename))
 568                 list_of_deferreds.append(d)
 569             deferList = defer.DeferredList(list_of_deferreds, consumeErrors=True)
 570             return deferList
 571         collective_dirmap_d.addCallback(scan_collective)
 572         def highest_version(deferredList):
 573             max_version = 0
 574             metadata = None
 575             node = None
 576             for success, result in deferredList:
 577                 if success:
 578                     if result[1]['version'] > max_version:
 579                         node, metadata = result
 580                         max_version = result[1]['version']
 581             return node, metadata
 582         collective_dirmap_d.addCallback(highest_version)
 583         return collective_dirmap_d
 584
 585     def _append_to_batch(self, name, file_node, metadata):
 586         if self._download_scan_batch.has_key(name):
 587             self._download_scan_batch[name] += [(file_node, metadata)]
 588         else:
 589             self._download_scan_batch[name] = [(file_node, metadata)]
 590
 591     def _scan_remote(self, nickname, dirnode):
 592         self._log("_scan_remote nickname %r" % (nickname,))
 593         d = dirnode.list()
 594         def scan_listing(listing_map):
 595             for encoded_relpath_u in listing_map.keys():
 596                 relpath_u = magicpath.magic2path(encoded_relpath_u)
 597                 self._log("found %r" % (relpath_u,))
 598
 599                 file_node, metadata = listing_map[encoded_relpath_u]
 600                 local_version = self._get_local_latest(relpath_u)
 601                 remote_version = metadata.get('version', None)
 602                 self._log("%r has local version %r, remote version %r" % (relpath_u, local_version, remote_version))
 603                 if local_version is None or remote_version is None or local_version < remote_version:
 604                     self._log("%r added to download queue" % (relpath_u,))
 605                     self._append_to_batch(relpath_u, file_node, metadata)
 606         d.addCallback(scan_listing)
 607         d.addBoth(self._logcb, "end of _scan_remote")
 608         return d
 609
 610     def _scan_remote_collective(self):
 611         self._log("_scan_remote_collective")
 612         self._download_scan_batch = {} # XXX
 613
 614         if self._collective_dirnode is None:
 615             return
 616         collective_dirmap_d = self._collective_dirnode.list()
 617         def do_list(result):
 618             others = [x for x in result.keys()]
 619             return result, others
 620         collective_dirmap_d.addCallback(do_list)
 621         def scan_collective(result):
 622             d = defer.succeed(None)
 623             collective_dirmap, others_list = result
 624             for dir_name in others_list:
 625                 d.addCallback(lambda x, dir_name=dir_name: self._scan_remote(dir_name, collective_dirmap[dir_name][0]))
 626                 # XXX todo add errback
 627             return d
 628         collective_dirmap_d.addCallback(scan_collective)
 629         collective_dirmap_d.addCallback(self._filter_scan_batch)
 630         collective_dirmap_d.addCallback(self._add_batch_to_download_queue)
 631         return collective_dirmap_d
 632
 633     def _add_batch_to_download_queue(self, result):
 634         self._log("result = %r" % (result,))
 635         self._log("deque = %r" % (self._deque,))
 636         self._deque.extend(result)
 637         self._log("deque after = %r" % (self._deque,))
 638         self._count('objects_queued', len(result))
 639         self._log("pending = %r" % (self._pending,))
 640         self._pending.update(map(lambda x: x[0], result))
 641         self._log("pending after = %r" % (self._pending,))
 642
 643     def _filter_scan_batch(self, result):
 644         self._log("_filter_scan_batch")
 645         extension = [] # consider whether this should be a dict
 646         for relpath_u in self._download_scan_batch.keys():
 647             if relpath_u in self._pending:
 648                 continue
 649             file_node, metadata = max(self._download_scan_batch[relpath_u], key=lambda x: x[1]['version'])
 650             if self._should_download(relpath_u, metadata['version']):
 651                 extension += [(relpath_u, file_node, metadata)]
 652             else:
 653                 self._log("Excluding %r" % (relpath_u,))
 654                 self._count('objects_excluded')
 655                 self._call_hook(None, 'processed')
 656         return extension
 657
 658     def _when_queue_is_empty(self):
 659         d = task.deferLater(self._clock, self._turn_delay, self._scan_remote_collective)
 660         d.addBoth(self._logcb, "after _scan_remote_collective")
 661         d.addCallback(lambda ign: self._turn_deque())
 662         return d
 663
 664     def _process(self, item, now=None):
 665         self._log("_process(%r)" % (item,))
 666         if now is None:
 667             now = time.time()
 668         (relpath_u, file_node, metadata) = item
 669         fp = self._get_filepath(relpath_u)
 670         abspath_u = unicode_from_filepath(fp)
 671         conflict_path_u = self._get_conflicted_filename(abspath_u)
 672         d = defer.succeed(None)
 673
 674         def do_update_db(written_abspath_u):
 675             filecap = file_node.get_uri()
 676             last_uploaded_uri = metadata.get('last_uploaded_uri', None)
 677             last_downloaded_uri = filecap
 678             last_downloaded_timestamp = now
 679             written_pathinfo = get_pathinfo(written_abspath_u)
 680
 681             if not written_pathinfo.exists and not metadata.get('deleted', False):
 682                 raise Exception("downloaded object %s disappeared" % quote_local_unicode_path(written_abspath_u))
 683
 684             self._db.did_upload_version(relpath_u, metadata['version'], last_uploaded_uri,
 685                                         last_downloaded_uri, last_downloaded_timestamp, written_pathinfo)
 686             self._count('objects_downloaded')
 687         def failed(f):
 688             self._log("download failed: %s" % (str(f),))
 689             self._count('objects_failed')
 690             return f
 691
 692         if os.path.isfile(conflict_path_u):
 693             def fail(res):
 694                 raise ConflictError("download failed: already conflicted: %r" % (relpath_u,))
 695             d.addCallback(fail)
 696         else:
 697             is_conflict = False
 698             if self._db.check_file_db_exists(relpath_u):
 699                 dmd_last_downloaded_uri = metadata.get('last_downloaded_uri', None)
 700                 local_last_downloaded_uri = self._db.get_last_downloaded_uri(relpath_u)
 701                 print "metadata %r" % (metadata,)
 702                 print "<<<<--- if %r != %r" % (dmd_last_downloaded_uri, local_last_downloaded_uri)
 703                 if dmd_last_downloaded_uri is not None and local_last_downloaded_uri is not None:
 704                     if dmd_last_downloaded_uri != local_last_downloaded_uri:
 705                         is_conflict = True
 706                         self._count('objects_conflicted')
 707
 708                 #dmd_last_uploaded_uri = metadata.get('last_uploaded_uri', None)
 709                 #local_last_uploaded_uri = ...
 710
 711             if relpath_u.endswith(u"/"):
 712                 if metadata.get('deleted', False):
 713                     self._log("rmdir(%r) ignored" % (abspath_u,))
 714                 else:
 715                     self._log("mkdir(%r)" % (abspath_u,))
 716                     d.addCallback(lambda ign: fileutil.make_dirs(abspath_u))
 717                     d.addCallback(lambda ign: abspath_u)
 718             else:
 719                 if metadata.get('deleted', False):
 720                     d.addCallback(lambda ign: self._rename_deleted_file(abspath_u))
 721                 else:
 722                     d.addCallback(lambda ign: file_node.download_best_version())
 723                     d.addCallback(lambda contents: self._write_downloaded_file(abspath_u, contents,
 724                                                                                is_conflict=is_conflict))
 725
 726         d.addCallbacks(do_update_db, failed)
 727
 728         def remove_from_pending(res):
 729             self._pending.remove(relpath_u)
 730             return res
 731         d.addBoth(remove_from_pending)
 732         def trap_conflicts(f):
 733             f.trap(ConflictError)
 734             return None
 735         d.addErrback(trap_conflicts)
 736         return d