class LibrarianCore(metaclass=API):
    ''' Base class for caching systems common to non-volatile librarians
    such as DiskLibrarian, S3Librarian, etc.
    
    TODO: make ghid vs frame ghid usage more consistent across things.
    
    NOTE: ideally, everything stateful should be in the librarian. In
    other words, you shouldn't use a bookie to track binding status and
    stuff; that should all be handled within librarian. That would make
    the librarian the single source of truth. Unfortunately, there's not
    really a good way of doing that right now ops-wise; you'd need some
    kind of relational filesystem.
    
    TODO: instead, should create a dedicated "cache" system. It can be
    responsible for restoration of stuff, atomic state updates, etc.
    '''
    _enforcer = weak_property('__enforcer')
    _lawyer = weak_property('__lawyer')
    _percore = weak_property('__percore')
    
    @public_api
    def __init__(self, *args, memory_cache=10000, **kwargs):
        super().__init__(*args, **kwargs)
        
        # Lookup for ghid -> hypergolix description
        # This may be GC'd by the python process.
        self._catalog = FiniteDict(maxlen=memory_cache)
        
    @__init__.fixture
    def __init__(self, *args, **kwargs):
        ''' Construct an in-memory-only version of librarian.
        '''
        super(LibrarianCore.__fixture__, self).__init__(*args, **kwargs)
        
        # Lookup <ghid>: <raw data>
        self._shelf = {}
        
        # Lookup for dynamic ghid -> frame ghid
        self._dyn_resolver = {}
        
        # Lookup <bound ghid>: set(<binding obj>)
        self._bound_by_ghid = SetMap()
        
        # Lookup <debound ghid>: set(<debinding ghid>)
        self._debound_by_ghid = SetMap()
        
        # Lookup <recipient>: set(<request ghid>)
        self._requests_for_recipient = SetMap()
        
    @fixture_api
    def RESET(self):
        ''' Reset all of the librarian.
        '''
        self._catalog.clear()
        self._shelf.clear()
        self._dyn_resolver.clear()
        self._bound_by_ghid.clear_all()
        self._debound_by_ghid.clear_all()
        self._requests_for_recipient.clear_all()
        
    def assemble(self, enforcer, lawyer, percore):
        ''' Assign stuff hereto avoid circuitous problems.
        '''
        # We need to be able to check the legality of debindings whose targets
        # are uploaded after the debinding itself is.
        self._enforcer = enforcer
        self._lawyer = lawyer
        # And we need to be able to load things back up from the cache
        self._percore = percore
    
    @public_api
    async def is_bound(self, obj):
        ''' Check to see if the object has been bound.
        '''
        bindings = await self.bind_status(obj.ghid)
        return bool(bindings)
    
    @public_api
    async def is_debound(self, obj):
        # We need to validate any unvalidated bindings. For now, just always
        # validate debindings when an object is being checked.
        debindings = await self.debind_status(obj.ghid)
        invalidated = 0
        for debinding_ghid in debindings:
            # Get the existing debinding object
            debinding = await self.summarize(debinding_ghid)
            
            # Validate existing binding against newly-known target
            try:
                await self._enforcer.validate_gdxx(debinding, target_obj=obj)
                await self._lawyer.validate_gdxx(debinding, target_obj=obj)
                
            # Validation failed. Remove illegal debinding.
            except (InvalidTarget, InconsistentAuthor):
                logger.error(''.join((
                    'Removed invalid existing debinding.\n',
                    '    Debinding author:     ', str(debinding.author), '\n',
                    '    Valid object author:  ', str(obj.author), '\n',
                    '    Debinding target:     ', str(debinding.target), '\n',
                    '    Target type:          ', str(type(obj))
                )))
                # There's no need to do anything with the undertaker, because
                # debindings don't get subscribed to, and we (the librarian)
                # are the sole source of truth.
                await self.abandon(debinding)
                
                # Record that this was invalidated
                invalidated += 1
                
            # If we were differentiating between validated debindings and
            # provisional (unvalidated) debindings, this would be the place to
            # commit the provisional to the validated.
        
        # Instead of making another call to debind_status, we can calculate it
        # directly. NOTE! There is a race condition between the initial
        # validity check on upload, and the commit call.
        
        return bool(len(debindings) - invalidated)
        
    @is_debound.fixture
    async def is_debound(self, obj):
        ''' Create ad-hoc enforcers and lawyers when fixturing self.
        '''
        enforcer = Enforcer.__fixture__(librarian=self)
        lawyer = LawyerCore.__fixture__(librarian=self)
        self._enforcer = enforcer
        self._lawyer = lawyer
        result = await super(LibrarianCore.__fixture__, self).is_debound(obj)
        del enforcer
        del lawyer
        return result
    
    @public_api
    async def store(self, obj, data):
        ''' Starts tracking an object.
        obj is a hypergolix representation object.
        raw is bytes-like.
        '''
        if isinstance(obj, _GobdLite):
            reference_ghid = obj.frame_ghid
            # If we have an existing frame, get ITS frame ghid, so we can clear
            # it if the storage succeeds.
            try:
                old_ghid = await self.resolve_frame(obj.ghid)
            except KeyError:
                old_ghid = None
            
        else:
            reference_ghid = obj.ghid
            old_ghid = None
            
        await self.add_to_cache(obj, data)
        self._catalog[reference_ghid] = obj
        
        # If successful (which is any time we get to here), we also need to get
        # rid of any old dynamic frames and pop them from the catalog.
        if old_ghid is not None:
            await self.remove_from_cache(old_ghid)
            # We need the None regardless of bugs, in case the old frame is
            # "stale" enough to have been released from memory
            self._catalog.pop(old_ghid, None)
    
    @public_api
    async def retrieve(self, ghid):
        ''' Returns the raw data associated with the ghid, checking only
        locally.
        '''
        try:
            ghid = await self.resolve_frame(ghid)
        except KeyError:
            pass
            
        return (await self.get_from_cache(ghid))
    
    @public_api
    async def summarize(self, ghid):
        ''' Returns a lightweight Hypergolix description of the object.
        Checks only locally.
        '''
        try:
            ghid = await self.resolve_frame(ghid)
        except KeyError:
            pass
        
        try:
            obj = self._catalog[ghid]
        
        except KeyError:
            logger.debug('Attempting lazy-load for ' + str(ghid))
            # This will raise DoesNotExist if missing.
            data = await self.get_from_cache(ghid)
            # This does NOT ingest the data into the persistence system!
            obj = await self._percore.attempt_load(data, quiet=False)
            self._catalog[ghid] = obj
            
        return obj
        
    @public_api
    async def abandon(self, obj):
        ''' Forces erasure of an object without notifying anyone else.
        Idempotent. Should never raise KeyError.
        '''
        if isinstance(obj, _GobdLite):
            ghid = obj.frame_ghid
        else:
            ghid = obj.ghid
        
        await self.remove_from_cache(ghid)
        
        # Delete it from the catalog (if it exists there)
        self._catalog.pop(ghid, None)
    
    # Subclasses MAY define this, but are not required to do so.
    @fixture_api
    async def restore(self):
        ''' For LibrarianCore, do nothing. This is just here as an
        endpoint for restore calls. If subclasses want/need to support
        restoration, they should override this.
        '''
    
    # Subclasses MUST define this to work!
    # @abc.abstractmethod
    @fixture_api
    async def contains(self, ghid):
        ''' Checks the ghidcache for the ghid.
        '''
        try:
            ghid = await self.resolve_frame(ghid)
        except KeyError:
            pass
        
        # Catalog may only be accurate locally. Shelf is accurate globally.
        return ghid in self._shelf
    
    # Subclasses MUST define this to work!
    # @abc.abstractmethod
    @fixture_api
    async def resolve_frame(self, ghid):
        ''' Get the current frame ghid from the dynamic ghid.
        '''
        if not isinstance(ghid, Ghid):
            raise TypeError('Ghid must be a Ghid.')
            
        if ghid in self._dyn_resolver:
            return self._dyn_resolver[ghid]
        else:
            raise KeyError(str(ghid) + ' not known as dynamic ghid.')
    
    # Subclasses MUST define this to work!
    # @abc.abstractmethod
    @fixture_api
    async def recipient_status(self, ghid):
        ''' Return a frozenset of ghids assigned to the passed ghid as
        a recipient.
        '''
        return self._requests_for_recipient.get_any(ghid)
    
    # Subclasses MUST define this to work!
    # @abc.abstractmethod
    @fixture_api
    async def bind_status(self, ghid):
        ''' Return a frozenset of ghids binding the passed ghid.
        '''
        return self._bound_by_ghid.get_any(ghid)
    
    # Subclasses MUST define this to work!
    # @abc.abstractmethod
    @fixture_api
    async def debind_status(self, ghid):
        ''' Return either a ghid, or None.
        '''
        # Note that any particular object can have exactly zero or one VALID
        # debinds, but that a malicious actor could find a race condition and
        # debind something FOR SOMEONE ELSE before the bookie knows about the
        # original object authorship.
        return self._debound_by_ghid.get_any(ghid)
            
    # Subclasses MUST define this to work!
    # @abc.abstractmethod
    @fixture_api
    async def add_to_cache(self, obj, data):
        ''' Adds the passed raw data to the cache.
        '''
        if isinstance(obj, _GobsLite):
            reference_ghid = obj.ghid
            self._bound_by_ghid.add(obj.target, obj.ghid)
            
        elif isinstance(obj, _GobdLite):
            reference_ghid = obj.frame_ghid
            try:
                existing = await self.summarize(obj.ghid)
            except KeyError:
                pass
            else:
                self._bound_by_ghid.remove(existing.target, obj.ghid)
                
            # Now we have a clean slate and need to update things accordingly.
            self._bound_by_ghid.add(obj.target, obj.ghid)
            self._dyn_resolver[obj.ghid] = obj.frame_ghid
                
        elif isinstance(obj, _GdxxLite):
            reference_ghid = obj.ghid
            self._debound_by_ghid.add(obj.target, obj.ghid)
            
        elif isinstance(obj, _GarqLite):
            reference_ghid = obj.ghid
            # NOTE: this is only necessary for a persistence server
            self._requests_for_recipient.add(obj.recipient, obj.ghid)
            
        else:
            reference_ghid = obj.ghid
            
        self._shelf[reference_ghid] = data
        
    # Subclasses MUST define this to work!
    # @abc.abstractmethod
    @fixture_api
    async def remove_from_cache(self, ghid):
        ''' Removes the data associated with the passed ghid from the
        cache.
        '''
        obj = await self.summarize(ghid)
        
        if isinstance(obj, _GobsLite):
            reference_ghid = obj.ghid
            self._bound_by_ghid.discard(obj.target, obj.ghid)
            
        elif isinstance(obj, _GobdLite):
            reference_ghid = obj.frame_ghid
            self._bound_by_ghid.discard(obj.target, obj.ghid)
            
            if self._dyn_resolver.get(obj.ghid, None) == obj.frame_ghid:
                self._dyn_resolver.pop(obj.ghid, None)
            
        elif isinstance(obj, _GdxxLite):
            reference_ghid = obj.ghid
            self._debound_by_ghid.discard(obj.target, obj.ghid)
            
        elif isinstance(obj, _GarqLite):
            reference_ghid = obj.ghid
            self._requests_for_recipient.discard(obj.recipient, obj.ghid)
            
        else:
            reference_ghid = obj.ghid
        
        try:
            self._shelf.pop(reference_ghid)
        except KeyError as exc:
            # Protect the full GHID from accidental exposure
            raise DoesNotExist(str(ghid)) from None
        
    # Subclasses MUST define this to work!
    # @abc.abstractmethod
    @fixture_api
    async def get_from_cache(self, ghid):
        ''' Returns the raw data associated with the ghid.
        '''
        try:
            return self._shelf[ghid]
        except KeyError as exc:
            # Protect the full GHID from accidental exposure
            raise DoesNotExist(str(ghid)) from None class PostalCore(loopa.TaskLooper, metaclass=API):
    ''' Tracks, delivers notifications about objects using **only weak
    references** to them. Threadsafe.
    
    ♫ Please Mister Postman... ♫
    
    Question: should the distributed state management of GARQ recipients
    be managed here, or in the bookie (where it currently is)?
    '''
    _librarian = weak_property('__librarian')
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        
        # The scheduling queue is created at loop init.
        self._scheduled = None
        # The delayed lookup. <awaiting ghid>: set(<subscribed ghids>)
        self._deferred = SetMap()
        
        # Resolve primitives into their schedulers.
        self._scheduler_lookup = {
            _GidcLite: self._schedule_gidc,
            _GeocLite: self._schedule_geoc,
            _GobsLite: self._schedule_gobs,
            _GobdLite: self._schedule_gobd,
            _GdxxLite: self._schedule_gdxx,
            _GarqLite: self._schedule_garq
        }
        
    @fixture_api
    def RESET(self):
        ''' In general this would be where you'd reset self._scheduled,
        but since self.schedule() is fixtured as a NOOP, there's no
        real reason to do anything here.
        '''
        
    def assemble(self, librarian):
        # Links the librarian.
        self._librarian = librarian
        
    async def await_idle(self):
        ''' Wait until the postman has no more deliveries to perform.
        '''
        while self._scheduled is None:
            await asyncio.sleep(.01)
        
        await self._scheduled.join()
        
    async def loop_init(self):
        ''' Init all of the needed async primitives.
        '''
        self._scheduled = asyncio.Queue()
        
    async def loop_run(self):
        ''' Deliver notifications as soon as they are available.
        TODO: support parallel sending.
        '''
        subscription, notification, skip_conn = await self._scheduled.get()
        
        try:
            logger.info(str(subscription) + ' subscription out for delivery.')
            # We can't spin this out into a thread because some of our
            # delivery mechanisms want this to have an event loop.
            await self._deliver(subscription, notification, skip_conn)
            
        except asyncio.CancelledError:
            logger.debug('PostalCore cancelled.')
            raise
        
        except Exception:
            logger.error(str(subscription) + ' subscription FAILED for ' +
                         'notification ' + str(notification) +
                         ' w/ traceback:\n' +
                         ''.join(traceback.format_exc()))
            
        finally:
            self._scheduled.task_done()
        
    async def loop_stop(self):
        ''' Clear the async primitives.
        '''
        # Ehhhhh, should the queue be emptied before being destroyed?
        self._scheduled = None
        
    @fixture_return(True)
    @public_api
    async def schedule(self, obj, removed=False, skip_conn=None):
        ''' Schedules update delivery for the passed object.
        '''
        # It's possible we're being told to schedule nothing, so catch that
        # here.
        if obj is None:
            return
        
        else:
            try:
                scheduler = self._scheduler_lookup[type(obj)]
            
            except KeyError:
                raise TypeError('Postman scheduling failed for ' +
                                type(obj).__name__) from None
            
            else:
                await scheduler(obj, removed, skip_conn)
                
            return True
        
    async def _schedule_gidc(self, obj, removed, skip_conn):
        # GIDC will never trigger a subscription.
        pass
        
    async def _schedule_geoc(self, obj, removed, skip_conn):
        # GEOC will never trigger a subscription directly, though they might
        # have deferred updates.
        # Note that these have already been put into _SubsUpdate form.
        for deferred in self._deferred.pop_any(obj.ghid):
            await self._scheduled.put(deferred)
        
    async def _schedule_gobs(self, obj, removed, skip_conn):
        # GOBS will never trigger a subscription.
        pass
        
    async def _schedule_gobd(self, obj, removed, skip_conn):
        # GOBD might trigger a subscription! But, we also might to need to
        # defer it. Or, we might be removing it.
        if removed:
            debinding_ghids = await self._librarian.debind_status(obj.ghid)
            
            # Check to see that there are the proper number of debindings
            num_debindings = len(debinding_ghids)
            if num_debindings != 1:
                raise ValueError('Improper debinding number: ' +
                                 str(num_debindings) + ' for ' + str(obj))
                
            # Debinding_ghids is a frozenset; this is the fastest way of
            # getting the single element from it.
            debinding_ghid = next(iter(debinding_ghids))
            await self._scheduled.put(
                _SubsUpdate(obj.ghid, debinding_ghid, skip_conn)
            )
            
        else:
            notifier = _SubsUpdate(obj.ghid, obj.frame_ghid, skip_conn)
            if (await self._librarian.contains(obj.target)):
                logger.debug(str(obj) +
                             ' subscription notification scheduled for: ' +
                             str(obj.target))
                await self._scheduled.put(notifier)
            
            else:
                self._deferred.add(obj.target, notifier)
                logger.debug(str(obj) +
                             ' subscription notification deferred for: ' +
                             str(obj.target))
        
    async def _schedule_gdxx(self, obj, removed, skip_conn):
        # GDXX will never directly trigger a subscription. If they are removing
        # a subscribed object, the actual removal (in the undertaker GC) will
        # trigger a subscription without us.
        pass
        
    async def _schedule_garq(self, obj, removed, skip_conn):
        # GARQ might trigger a subscription! Or we might be removing it.
        if removed:
            debinding_ghids = await self._librarian.debind_status(obj.ghid)
            
            # Check to see that there are the proper number of debindings
            num_debindings = len(debinding_ghids)
            if num_debindings != 1:
                raise ValueError('Improper debinding number: ' +
                                 str(num_debindings) + ' for ' + str(obj))
                
            # Debinding_ghids is a frozenset; this is the fastest way of
            # getting the single element from it.
            debinding_ghid = next(iter(debinding_ghids))
            await self._scheduled.put(
                _SubsUpdate(obj.recipient, debinding_ghid, skip_conn)
            )
        else:
            await self._scheduled.put(
                _SubsUpdate(obj.recipient, obj.ghid, skip_conn)
            )
            
    async def _deliver(self, subscription, notification, skip_conn):
        ''' Do the actual subscription update.
        '''
        # We need to freeze the listeners before we operate on them, but we
        # don't need to lock them while we go through all of the callbacks.
        # Instead, just sacrifice any subs being added concurrently to the
        # current delivery run.
        pass class PostOffice(PostalCore, metaclass=API):
    ''' Postman to use for remote persistence servers.
    '''
    _remote_protocol = weak_property('__remote_protocol')
    
    def __init__(self, *args, subs_timeout=30, **kwargs):
        super().__init__(*args, **kwargs)
        # By using WeakSetMap we can automatically handle dropped connections
        # Lookup <subscribed ghid>: set(<subscribed callbacks>)
        self._connections = WeakSetMap()
        self._subscriptions = WeakKeySetMap()
        
        self._subs_timeout = subs_timeout
        
    def assemble(self, librarian, remote_protocol):
        super().assemble(librarian)
        self._remote_protocol = remote_protocol
    
    @fixture_noop
    @public_api
    async def subscribe(self, connection, ghid):
        ''' Tells the postman that the connection would like to be
        updated about ghid.
        '''
        logger.debug('CONN ' + str(connection) + ' subscribed to ' + str(ghid))
        
        # First add the subscription listeners
        self._connections.add(ghid, connection)
        self._subscriptions.add(connection, ghid)
        
        # Now manually reinstate any desired notifications for garq requests
        # that have yet to be handled
        for existing_mail in (await self._librarian.recipient_status(ghid)):
            obj = await self._librarian.summarize(existing_mail)
            await self.schedule(obj)
    
    @fixture_noop
    @public_api
    async def unsubscribe(self, connection, ghid):
        ''' Remove the callback for ghid. Indempotent; will never raise
        a keyerror.
        '''
        try:
            # Discard the subscription (no keyerror)
            self._subscriptions.discard(connection, ghid)
            # Remove the connection (allowing it to raise if missing)
            self._connections.remove(ghid, connection)
            
        except KeyError:
            logger.debug('CONN ' + str(connection) + ' never subscribed to ' +
                         str(ghid))
            return False
            
        else:
            logger.debug('CONN ' + str(connection) + ' unsubscribed to ' +
                         str(ghid))
            return True
            
    async def _deliver(self, subscription, notification, skip_conn):
        ''' Do the actual subscription update.
        
        NOTE THAT SKIP_CONN is a weakref.ref.
        '''
        pkg_label = (str(subscription) + ' subscription, notification ' +
                     str(notification))
        # We need to freeze the listeners before we operate on them, but we
        # don't need to lock them while we go through all of the callbacks.
        # Instead, just sacrifice any subs being added concurrently to the
        # current delivery run.
        connections = self._connections.get_any(subscription)
        logger.debug(
            pkg_label + ' has ' + str(len(connections)) + ' listeners.'
        )
        
        # Resolve the weak reference to the connection
        if skip_conn is not None:
            # This could still be None, but that won't affect our comparison
            skip_conn = skip_conn()
            
        for connection in connections:
            if connection is not skip_conn:
                # Make this a background task, or one blocking connection can
                # hold up the entire subscription queue
                make_background_future(
                    self._remote_protocol.subscription_update(
                        connection,
                        subscription,
                        notification,
                        timeout = self._subs_timeout
                    )
                )
            else:
                logger.debug(pkg_label + ' skipped one connection.')
    
    @fixture_return(frozenset())
    @public_api
    async def list_subs(self, connection):
        ''' List all subscriptions for the connection.
        '''
        return self._subscriptions.get_any(connection)
    
    @fixture_noop
    @public_api
    async def clear_subs(self, connection):
        ''' Clear all subscriptions for the connection.
        '''
        subscriptions = self._subscriptions.pop_any(connection)
        for ghid in subscriptions:
            self._connections.discard(ghid, connection) class MrPostman(PostalCore, metaclass=API):
    ''' Postman to use for LOCAL persistence systems -- ie ones that
    have logins and do not support subsription.
    
    Note that MrPostman doesn't need to worry about silencing updates,
    because the persistence ingestion tract will only result in a mail
    run if there's a new object there. So, by definition, any re-sent
    objects will be DOA.
    '''
    _rolodex = weak_property('__rolodex')
    _golcore = weak_property('__golcore')
    _oracle = weak_property('__oracle')
    _salmonator = weak_property('__salmonator')
        
    def assemble(self, golcore, oracle, librarian, rolodex, salmonator):
        super().assemble(librarian)
        self._golcore = golcore
        self._rolodex = rolodex
        self._oracle = oracle
        self._salmonator = salmonator
            
    async def _deliver(self, subscription, notification, skip_conn):
        ''' Do the actual subscription update.
        
        NOTE THAT SKIP_CONN is a weakref.ref (or None).
        '''
        # We just got a garq for our identity. Rolodex handles these.
        if subscription == self._golcore.whoami:
            logger.debug(str(subscription) +
                         'subscription notification handed off to rolodex...')
            await self._rolodex.notification_handler(
                subscription,
                notification
            )
        
        # Anything else is an object subscription. Handle those by directly,
        # but only if we have them in memory.
        # elif subscription in self._oracle:
        # Note: that was causing issues so I'm holding off on it as premature
        # optimization / feature creep
        else:
            # The ingestion pipeline will already have applied any new updates
            # to the ghidproxy.
            try:
                obj = await self._oracle.get_object(GAOCore, subscription)
            
            except KeyError:
                logger.debug(str(subscription) + ' subscription delivery ' +
                             'ignored: unavailable at oracle: ' +
                             str(notification))
                await self._salmonator.deregister(subscription)
                
            else:
                logger.debug(str(subscription) + ' subscription delivery ' +
                             'starting for ' + str(notification))
                await obj.pull(notification) class GolixCore(metaclass=API):
    ''' Wrapper around Golix library that automates much of the state
    management, holds the Agent's identity, etc etc.
    '''
    _librarian = weak_property('__librarian')
    DEFAULT_LEGROOM = 7
    
    # Async stuff
    _executor = readonly_property('__executor')
    _loop = readonly_property('__loop')
    
    @public_api
    def __init__(self, executor, loop, *args, **kwargs):
        ''' Create a new agent. Persister should subclass _PersisterBase
        (eventually this requirement may be changed).
        
        persister isinstance _PersisterBase
        dispatcher isinstance DispatcherBase
        _identity isinstance golix.FirstParty
        '''
        super().__init__(*args, **kwargs)
        self._mutex_request = threading.Lock()
        self._mutex_container = threading.Lock()
        self._mutex_sbinding = threading.Lock()
        self._mutex_dbinding = threading.Lock()
        self._mutex_xbinding = threading.Lock()
        
        # Added during bootstrap
        self.__identity = None
        
        # Async-specific stuff
        setattr(self, '__executor', executor)
        setattr(self, '__loop', loop)
        
    @__init__.fixture
    def __init__(self, test_agent, *args, librarian=None, **kwargs):
        ''' Just, yknow, throw in the towel. Err, agent. Whatever.
        '''
        super(GolixCore.__fixture__, self).__init__(
            executor = None,
            loop = None,
            *args,
            **kwargs
        )
        self._identity = test_agent
        
        if librarian is not None:
            self._librarian = librarian
        
    def assemble(self, librarian):
        # Chicken, meet egg.
        self._librarian = librarian
        
    def bootstrap(self, account):
        # This must be done ASAGDFP. Must be absolute first thing to bootstrap.
        self._identity = account._identity
        
    @property
    def _identity(self):
        ''' Wrap our identity in such a way that, if it's not defined,
        we can fall back on the FirstParty class.
        '''
        if self.__identity is None:
            return FirstParty
        else:
            return self.__identity
            
    @_identity.setter
    def _identity(self, identity):
        ''' Set the identity.
        '''
        self.__identity = identity
        
    @property
    @public_api
    def whoami(self):
        ''' Return the Agent's Ghid.
        '''
        return self._identity.ghid
    
    @public_api
    async def unpack_request(self, request):
        ''' Just like it says on the label...
        Note that the request is PACKED, not unpacked.
        '''
        # Run the actual function in the executor
        return (await self._loop.run_in_executor(
            self._executor,
            self._unpack_request,
            request
        ))
        
    @unpack_request.fixture
    async def unpack_request(self, request):
        ''' Bypass executor for the fixture.
        '''
        return self._unpack_request(request)
        
    def _unpack_request(self, request):
        ''' Just like it says on the label...
        Note that the request is PACKED, not unpacked.
        '''
        with self._mutex_request:
            unpacked = self._identity.unpack_request(request)
        return unpacked
        
    @public_api
    async def open_request(self, unpacked):
        ''' Just like it says on the label...
        Note that the request is UNPACKED, not packed.
        '''
        try:
            requestor = SecondParty.from_packed(
                await self._librarian.retrieve(unpacked.author)
            )
            
        except KeyError as exc:
            raise UnknownParty(
                'Request author unknown: ' + str(unpacked.author)
            ) from exc
        
        # Run the actual function in the executor
        return (await self._loop.run_in_executor(
            self._executor,
            self._open_request,
            unpacked,
            requestor
        ))
        
    @open_request.fixture
    async def open_request(self, unpacked):
        ''' Also bypass executor here.
        '''
        requestor = SecondParty.from_packed(
            await self._librarian.retrieve(unpacked.author)
        )
        return self._open_request(unpacked, requestor)
        
    def _open_request(self, unpacked, requestor):
        ''' Just like it says on the label...
        Note that the request is UNPACKED, not packed.
        '''
        return self._identity.receive_request(requestor, unpacked)
    
    @public_api
    async def make_request(self, recipient, payload):
        # Just like it says on the label...
        try:
            recipient = SecondParty.from_packed(
                await self._librarian.retrieve(recipient)
            )
        except KeyError as exc:
            raise UnknownParty(
                'Request author unknown: ' + str(recipient)
            ) from exc
        
        # Run the actual function in the executor
        return (await self._loop.run_in_executor(
            self._executor,
            self._make_request,
            recipient,
            payload
        ))
        
    @make_request.fixture
    async def make_request(self, recipient, payload):
        ''' Bypass the goddamn executor. Ffs.
        '''
        recipient = SecondParty.from_packed(
            await self._librarian.retrieve(recipient)
        )
        return self._make_request(recipient, payload)
        
    def _make_request(self, recipient, payload):
        # Just like it says on the label...
        with self._mutex_request:
            return self._identity.make_request(
                recipient = recipient,
                request = payload,
            )
    
    @public_api
    async def open_container(self, container, secret):
        author = SecondParty.from_packed(
            await self._librarian.retrieve(container.author)
        )
        
        # Wrapper around golix.FirstParty.receive_container.
        # Run the actual function in the executor
        return (await self._loop.run_in_executor(
            self._executor,
            self._open_container,
            container,
            secret,
            author
        ))
        
    @open_container.fixture
    async def open_container(self, container, secret):
        ''' Bypass executor for fixture.
        '''
        author = SecondParty.from_packed(
            await self._librarian.retrieve(container.author)
        )
        
        return self._open_container(container, secret, author)
        
    def _open_container(self, container, secret, author):
        # Wrapper around golix.FirstParty.receive_container.
        with self._mutex_container:
            return self._identity.receive_container(
                author = author,
                secret = secret,
                container = container
            )
    
    @public_api
    async def make_container(self, data, secret):
        # Simple wrapper around golix.FirstParty.make_container
        # Run the actual function in the executor
        return (await self._loop.run_in_executor(
            self._executor,
            self._make_container,
            data,
            secret
        ))
        
    @make_container.fixture
    async def make_container(self, data, secret):
        ''' Bypass executor.
        '''
        return self._make_container(data, secret)
        
    def _make_container(self, data, secret):
        # Simple wrapper around golix.FirstParty.make_container
        with self._mutex_container:
            return self._identity.make_container(
                secret = secret,
                plaintext = data
            )

    @public_api
    async def make_binding_stat(self, target):
        # Note that this requires no open() method, as bindings are verified by
        # the local persister.
        # Run the actual function in the executor
        return (await self._loop.run_in_executor(
            self._executor,
            self._make_binding_stat,
            target
        ))
        
    @make_binding_stat.fixture
    async def make_binding_stat(self, target):
        ''' Skip the executor for the fixture.
        '''
        return self._make_binding_stat(target)

    def _make_binding_stat(self, target):
        # Note that this requires no open() method, as bindings are verified by
        # the local persister.
        with self._mutex_sbinding:
            return self._identity.make_bind_static(target)
    
    @public_api
    async def make_binding_dyn(self, target, ghid=None, history=None):
        ''' Make a new dynamic binding frame.
        If supplied, ghid is the dynamic address, and history is an
        ordered iterable of the previous frame ghids.
        '''
        # Either ghid AND history must be defined, XOR ghid AND history must be
        # undefined.
        if bool(ghid) ^ bool(history):
            raise ValueError('Mixed def of ghid/history while dyn binding.')
        
        if ghid:
            current_binding = await self._librarian.summarize(ghid)
            counter = current_binding.counter + 1
        else:
            counter = 0
            
        # Run the actual function in the executor
        return (await self._loop.run_in_executor(
            self._executor,
            self._make_binding_dyn,
            ghid,
            counter,
            target,
            history
        ))
        
    @make_binding_dyn.fixture
    async def make_binding_dyn(self, target, ghid=None, history=None):
        ''' Yep, just bypass the executor
        '''
        if ghid:
            current_binding = await self._librarian.summarize(ghid)
            counter = current_binding.counter + 1
        else:
            counter = 0
        
        return self._make_binding_dyn(ghid, counter, target, history)
        
    def _make_binding_dyn(self, ghid, counter, target, history):
        ''' Make a new dynamic binding frame.
        If supplied, ghid is the dynamic address, and history is an
        ordered iterable of the previous frame ghids.
        '''
        if history is not None:
            target_vector = [target, *history]
        else:
            target_vector = [target]
        
        with self._mutex_dbinding:
            return self._identity.make_bind_dynamic(
                counter = counter,
                target_vector = target_vector,
                ghid_dynamic = ghid
            )
    
    @public_api
    async def make_debinding(self, target):
        # Simple wrapper around golix.FirstParty.make_debind
        # Run the actual function in the executor
        return (await self._loop.run_in_executor(
            self._executor,
            self._make_debinding,
            target
        ))
        
    @make_debinding.fixture
    async def make_debinding(self, target):
        ''' Bypass executor for the fixture.
        '''
        return self._make_debinding(target)
        
    def _make_debinding(self, target):
        # Simple wrapper around golix.FirstParty.make_debind
        with self._mutex_xbinding:
            return self._identity.make_debind(target) class Rolodex(metaclass=API):
    ''' Handles sharing, requests, etc.
    
    In the future, will maintain a contacts list to approve/reject
    incoming requests. In the further future, will maintain sharing
    pipelines, negotiated through handshakes, to perform sharing
    symmetrically between contacts.
    '''
    _account = weak_property('__account')
    _golcore = weak_property('__golcore')
    _ghidproxy = weak_property('__ghidproxy')
    _privateer = weak_property('__privateer')
    _percore = weak_property('__percore')
    _librarian = weak_property('__librarian')
    _salmonator = weak_property('__salmonator')
    _dispatch = weak_property('__dispatch')
    
    @public_api
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        
        # Persistent dict-like lookup for
        # request_ghid -> (request_target, request recipient)
        self._pending_requests = None
        # Lookup for <target_obj_ghid, recipient> -> set(<app_tokens>)
        self._outstanding_shares = None
        
    @__init__.fixture
    def __init__(self, *args, **kwargs):
        ''' Init the fixture.
        '''
        super(Rolodex.__fixture__, self).__init__(*args, **kwargs)
        
        self.shared = {}
        self._pending_requests = {}
        self._outstanding_shares = {}
        
    @fixture_api
    def RESET(self):
        ''' Super simple reset alias to __init__.
        '''
        self.shared.clear()
        self._pending_requests.clear()
        self._outstanding_shares.clear()
        
    @fixture_noop
    @public_api
    def bootstrap(self, account):
        ''' Initialize distributed state.
        '''
        self._account = account
        
        # Persistent dict-like lookup for
        # request_ghid -> (request_target, request recipient)
        self._pending_requests = account.rolodex_pending
        
        # These need to be distributed but aren't yet. TODO!
        # Lookup for <target_obj_ghid, recipient> -> set(<app_tokens>)
        self._outstanding_shares = account.rolodex_outstanding
        
    def assemble(self, golcore, ghidproxy, privateer, percore, librarian,
                 salmonator, dispatch):
        # Chicken, meet egg.
        self._golcore = golcore
        self._ghidproxy = ghidproxy
        self._privateer = privateer
        self._percore = percore
        self._librarian = librarian
        self._salmonator = salmonator
        self._dispatch = dispatch
        
    @public_api
    async def share_object(self, target, recipient, requesting_token):
        ''' Share a target ghid with the recipient.
        '''
        if not isinstance(target, Ghid):
            raise TypeError(
                'target must be Ghid or similar.'
            )
        if not isinstance(recipient, Ghid):
            raise TypeError(
                'recipient must be Ghid or similar.'
            )

        sharepair = _SharePair(target, recipient)
            
        # For now, this is just doing a handshake with some typechecking.
        await self._hand_object(*sharepair)
        
        if requesting_token is not None:
            self._outstanding_shares.add(sharepair, requesting_token)
            await self._account.flush()
            
    @share_object.fixture
    async def share_object(self, target, recipient, requesting_token):
        ''' Pretty basic.
        '''
        # Ignore requesting token, because we won't generally set it when
        # testing this.
        self.shared[target] = recipient
        
    async def _hand_object(self, target, recipient):
        ''' Initiates a handshake request with the recipient.
        '''
        if not (await self._librarian.contains(recipient)):
            await self._salmonator.attempt_pull(recipient, quiet=True)
        
        recipient_identity = \
            (await self._librarian.summarize(recipient)).identity
        
        # This is guaranteed to resolve the container fully.
        container_ghid = await self._ghidproxy.resolve(target)
        
        # TODO: fix Golix library so this isn't such a shitshow re:
        # breaking abstraction barriers.
        handshake = self._golcore._identity.make_handshake(
            target = target,
            secret = self._privateer.get(container_ghid)
        )
        
        request = self._golcore._identity.make_request(
            recipient = recipient_identity,
            request = handshake
        )
        
        # Note that this must be called before publishing to the persister, or
        # there's a race condition between them.
        self._pending_requests[request.ghid] = target, recipient
        
        # Note the potential race condition here. Should catch errors with the
        # persister in case we need to resolve pending requests that didn't
        # successfully post.
        await self._percore.direct_ingest(
            obj = _GarqLite.from_golix(request),
            packed = request.packed,
            remotable = True
        )
    
    @fixture_noop
    @public_api
    async def notification_handler(self, subscription, notification):
        ''' Callback to handle any requests.
        '''
        if not (await self._librarian.contains(notification)):
            await self._salmonator.attempt_pull(notification, quiet=True)
        
        # Note that the notification could also be a GDXX.
        request_or_debind = await self._librarian.summarize(notification)
        
        if isinstance(request_or_debind, _GarqLite):
            await self._handle_request(notification)
            
        elif isinstance(request_or_debind, _GdxxLite):
            # This case should only be relevant if we have multiple agents
            # logged in at separate locations at the same time, processing the
            # same GARQs.
            await self._handle_debinding(request_or_debind)
            
        else:
            raise RuntimeError(
                'Unexpected Golix primitive while listening for requests.'
            )
            
    async def _handle_request(self, notification):
        ''' The notification is an asymmetric request. Deal with it.
        '''
        try:
            packed = await self._librarian.retrieve(notification)
            unpacked = await self._golcore.unpack_request(packed)
        
            # TODO: have this filter based on contacts.
            if not (await self._librarian.contains(unpacked.author)):
                await self._salmonator.attempt_pull(
                    unpacked.author,
                    quiet = True
                )
        
            payload = await self._golcore.open_request(unpacked)
            await self._dispatch_payload(payload, notification)

        # Don't forget to (always) debind.
        finally:
            debinding = await self._golcore.make_debinding(notification)
            await self._percore.direct_ingest(
                obj = _GdxxLite.from_golix(debinding),
                packed = debinding.packed,
                remotable = True
            )
        
    async def _handle_debinding(self, debinding):
        ''' The notification is a debinding. Deal with it.
        '''
        # For now we just need to remove any pending requests for the
        # debinding's target.
        try:
            del self._pending_requests[debinding.target]
            await self._account.flush()
        except KeyError:
            logger.debug(
                'Request at ' + str(debinding.ghid) + ' target missing ' +
                'from pending requests: ' + str(debinding.target)
            )
        
    async def _dispatch_payload(self, payload, source_ghid):
        ''' Appropriately handles a request payload.
        '''
        if isinstance(payload, AsymHandshake):
            logger.debug(
                'Share handshake received from ' + str(payload.author) +
                ' at ' + str(source_ghid) + ' for ' + str(payload.target)
            )
            await self._handle_handshake(payload, source_ghid)
            
        elif isinstance(payload, AsymAck):
            logger.debug(
                'Handshake ACK received from ' + str(payload.author) +
                ' at ' + str(source_ghid) + ' for ' + str(payload.target)
            )
            await self._handle_ack(payload)
            
        elif isinstance(payload, AsymNak):
            logger.debug(
                'Handshake NAK received from ' + str(payload.author) +
                ' at ' + str(source_ghid) + ' for ' + str(payload.target)
            )
            await self._handle_nak(payload)
            
        else:
            raise RuntimeError('Encountered an unknown request type.')
            
    async def _handle_handshake(self, request, source_ghid):
        ''' Handles a handshake request after reception.
        '''
        try:
            # First, we need to figure out what the actual container object's
            # address is, and then stage the secret for it.
            # ALWAYS attempt to pull it first, or we may have a stale frame
            # locally, and attempt to stage a new secret for a long-expired
            # frame
            await self._salmonator.attempt_pull(
                request.target,
                quiet = True
            )
            
            binding_or_obj = await self._librarian.summarize(request.target)
            if isinstance(binding_or_obj, _GobdLite):
                container_ghid = binding_or_obj.target
                self._privateer.quarantine(
                    ghid = container_ghid,
                    secret = request.secret
                )
                if not (await self._librarian.contains(container_ghid)):
                    await self._salmonator.attempt_pull(
                        container_ghid,
                        quiet = True
                    )
            elif isinstance(binding_or_obj, _GeocLite):
                self._privateer.quarantine(
                    ghid = binding_or_obj.ghid,
                    secret = request.secret
                )
            else:
                raise TypeError(
                    'Invalid handshake target: ' + str(type(binding_or_obj))
                )
            
            # Note that unless we raise a HandshakeError RIGHT NOW, we'll be
            # sending an ack to the handshake, just to indicate successful
            # receipt of the share. If the originating app wants to check for
            # availability, well, that's currently on them. In the future, add
            # handle for that in SHARE instead of HANDSHAKE?
            
        except asyncio.CancelledError:
            logger.debug('Handshake handler cancelled.')
            raise
            
        except Exception as exc:
            logger.error(
                'Exception encountered while handling a handshake. Returned a '
                'NAK.\n' + ''.join(traceback.format_exc())
            )
            # Erfolglos. Send a nak to whomever sent the handshake
            response_obj = self._golcore._identity.make_nak(
                target = source_ghid
            )
            
        else:
            # Success. Send an ack to whomever sent the handshake
            response_obj = self._golcore._identity.make_ack(
                target = source_ghid
            )
        
        response = await self._golcore.make_request(
            request.author,
            response_obj
        )
        await self._percore.direct_ingest(
            obj = _GarqLite.from_golix(response),
            packed = response.packed,
            remotable = True
        )
            
        await self.share_handler(request.target, request.author)
            
    async def _handle_ack(self, request):
        ''' Handles a handshake ack after reception.
        '''
        try:
            target, recipient = self._pending_requests.pop(request.target)
        except KeyError:
            logger.error(
                'Received an ACK for an unknown origin: ' +
                str(request.target)
            )
        else:
            await self.receipt_ack_handler(target, recipient)
            
    async def _handle_nak(self, request):
        ''' Handles a handshake nak after reception.
        '''
        try:
            target, recipient = self._pending_requests.pop(request.target)
        except KeyError:
            logger.error(
                'Received a NAK for an unknown origin: ' +
                str(request.target)
            )
        else:
            await self.receipt_nak_handler(target, recipient)
    
    @fixture_noop
    @public_api
    async def share_handler(self, target, sender):
        ''' Incoming share targets (well, their ghids anyways) are
        forwarded to the _ipccore.
        
        Only separate from _handle_handshake right now because in the
        future, object sharing will be at least partly handled within
        its own dedicated rolodex pipeline.
        '''
        # Make sure we actually have the target
        if not (await self._librarian.contains(target)):
            await self._salmonator.attempt_pull(target, quiet=True)
        
        # Distribute the share in the background
        make_background_future(
            self._dispatch.distribute_share(target, sender)
        )
    
    @fixture_noop
    @public_api
    async def receipt_ack_handler(self, target, recipient):
        ''' Receives a share ack from the rolodex and passes it on to
        the application that requested the share.
        '''
        sharepair = _SharePair(target, recipient)
        tokens = self._outstanding_shares.pop_any(sharepair)
        await self._account.flush()
        
        # Distribute the share success in the background
        make_background_future(
            self._dispatch.distribute_share_success(
                target,
                recipient,
                tokens
            )
        )
    
    @fixture_noop
    @public_api
    async def receipt_nak_handler(self, target, recipient):
        ''' Receives a share nak from the rolodex and passes it on to
        the application that requested the share.
        '''
        sharepair = _SharePair(target, recipient)
        tokens = self._outstanding_shares.pop_any(sharepair)
        await self._account.flush()
        
        # Distribute the share failure in the background
        make_background_future(
            self._dispatch.distribute_share_failure(
                target,
                recipient,
                tokens
            )
        ) class Oracle(metaclass=API):
    ''' Source for total internal truth and state tracking of objects.
    
    Maintains <ghid>: <obj> lookup. Used by dispatchers to track obj
    state. Might eventually be used by AgentBase. Just a quick way to
    store and retrieve any objects based on an associated ghid.
    '''
    # These are actually used by the oracle itself
    _salmonator = weak_property('__salmonator')
    
    # These are only here to pass along to GAOs
    _golcore = weak_property('__golcore')
    _ghidproxy = weak_property('__ghidproxy')
    _privateer = weak_property('__privateer')
    _percore = weak_property('__percore')
    _librarian = weak_property('__librarian')
    
    def __init__(self, *args, **kwargs):
        ''' Sets up internal tracking.
        '''
        super().__init__(*args, **kwargs)
        self._lookup = {}
        
    @fixture_api
    def RESET(self):
        ''' Simply re-call init.
        '''
        self._lookup.clear()
        
    def assemble(self, golcore, ghidproxy, privateer, percore, librarian,
                 salmonator):
        # Chicken, meet egg.
        self._golcore = golcore
        self._ghidproxy = ghidproxy
        self._privateer = privateer
        self._percore = percore
        self._librarian = librarian
        self._salmonator = salmonator
        
    @fixture_api
    def add_object(self, ghid, obj):
        ''' Add an object to the fixture.
        '''
        self._lookup[ghid] = obj
            
    @public_api
    async def get_object(self, gaoclass, ghid, *args, **kwargs):
        ''' Get an object.
        '''
        if ghid in self._lookup:
            # TODO: this is bad, because we're suppresing any potential argsig
            # problems. We should fix the abstraction so that it doesn't break
            # assumptions like that.
            obj = self._lookup[ghid]
            logger.debug(
                'GAO ' + str(ghid) + ' already exists in Oracle memory as: ' +
                str(type(obj))
            )
            
            if not isinstance(obj, gaoclass):
                raise TypeError(
                    'Object has already been resolved, and is not the '
                    'correct GAO class.'
                )
            else:
                logger.debug(
                    'GAO ' + str(ghid) + ' acceptable for ' + str(gaoclass)
                )
                
            await obj._ctx.wait()
            if ghid not in self._lookup:
                raise RuntimeError('Contentious delete while getting object.')
        
        else:
            logger.info(
                'GAO ' + str(ghid) +
                ' not currently in Oracle memory. Attempting load as: ' +
                str(gaoclass)
            )
            
            # First create the actual GAO. We do not need to have the ghid
            # downloaded to do this -- object creation is just making a Python
            # object locally.
            obj = gaoclass(
                ghid,
                None,   # dynamic
                None,   # author
                7,      # legroom (will be overwritten by pull)
                *args,
                golcore = self._golcore,
                ghidproxy = self._ghidproxy,
                privateer = self._privateer,
                percore = self._percore,
                librarian = self._librarian,
                **kwargs
            )
            
            # TODO: fix leaky abstraction
            obj._ctx = asyncio.Event()
            
            # Always do this first to make sure we have the most recent version
            # in all subsequent calls, without a race condition.
            self._lookup[ghid] = obj
            
            try:
                # Now immediately subscribe to the object upstream, so that there
                # is no race condition getting updates
                await self._salmonator.register(ghid)
                # Add deregister as a finalizer, but don't call it atexit. TODO:
                # fix leaky abstraction
                finalizer = weakref.finalize(obj, self._salmonator._deregister,
                                             ghid)
                finalizer.atexit = False
                # Explicitly pull the object from salmonator to ensure we have
                # the newest version, and that it is available locally in
                # librarian if also available anywhere else. Note that
                # salmonator handles modal switching for dynamic/static. It
                # will also (by default, with skip_refresh=False) pull in any
                # updates that have accumulated in the meantime.
                await self._salmonator.attempt_pull(ghid, quiet=True)
                
                # Now actually fetch the object. This may KeyError if the
                # ghid is still unknown.
                await obj._pull()
            
            # Got an exception? Revert the lookup and reraise
            except Exception:
                del self._lookup[ghid]
                raise
                
            # We have to release other waiters regardless
            finally:
                # Finally, let future callers unblock
                obj._ctx.set()
            
        return obj
        
    @get_object.fixture
    async def get_object(self, gaoclass, ghid, *args, **kwargs):
        ''' Do the easy thing and just pull it out of lookup.
        '''
        # We should also check argsig while we're at it
        signature = inspect.Signature.from_callable(gaoclass)
        signature.bind(
            ghid,
            None,   # dynamic
            None,   # author
            7,      # legroom (will be overwritten by pull)
            *args,
            golcore = self,     # It's not going to be used and may not exist
            ghidproxy = self,   # It's not going to be used and may not exist
            privateer = self,   # It's not going to be used and may not exist
            percore = self,     # It's not going to be used and may not exist
            librarian = self,   # It's not going to be used and may not exist
            **kwargs
        )
            
        return self._lookup[ghid]
        
    @public_api
    async def new_object(self, gaoclass, dynamic, legroom, *args, **kwargs):
        ''' Creates a new object and returns it. Passes all *kwargs to
        the declared gao_class. Requires a zeroth state, and calls push
        internally.
        '''
        obj = gaoclass(
            None,                   # ghid
            dynamic,
            self._golcore.whoami,   # author
            legroom,
            *args,
            golcore = self._golcore,
            ghidproxy = self._ghidproxy,
            privateer = self._privateer,
            percore = self._percore,
            librarian = self._librarian,
            **kwargs
        )
        await obj._push()
        
        # This is used strictly by oracle. TODO: fix leaky abstraction
        obj._ctx = asyncio.Event()
        obj._ctx.set()
        # Do this before registering with salmonator, in case the latter errors
        self._lookup[obj.ghid] = obj
        
        # Finally, register to receive any concurrent updates from other
        # simultaneous sessions, and then return the object
        await self._salmonator.register(obj.ghid)
        # Add deregister as a finalizer, but don't call it atexit. TODO: fix
        # leaky abstraction
        finalizer = weakref.finalize(obj, self._salmonator._deregister,
                                     obj.ghid)
        finalizer.atexit = False
        
        return obj
            
    @new_object.fixture
    async def new_object(self, gaoclass, dynamic, legroom, *args, **kwargs):
        ''' Relies upon add_object, but otherwise just pops something
        from the lookup.
        '''
        # We should also check argsig while we're at it
        signature = inspect.Signature.from_callable(gaoclass)
        signature.bind(
            None,               # ghid
            dynamic,
            None,               # author (normally assigned through whoami)
            legroom,
            *args,
            golcore = self,     # It's not going to be used and may not exist
            ghidproxy = self,   # It's not going to be used and may not exist
            privateer = self,   # It's not going to be used and may not exist
            percore = self,     # It's not going to be used and may not exist
            librarian = self,   # It's not going to be used and may not exist
            **kwargs
        )
        
        ghid, obj = self._lookup.popitem()
        self._lookup[ghid] = obj
        return obj
        
    def forget(self, ghid):
        ''' Removes the object from the cache. Next time an application
        wants it, it will need to be acquired from persisters.
        
        Indempotent; will not raise KeyError if called more than once.
        '''
        try:
            del self._lookup[ghid]
        except KeyError:
            logger.debug(str(ghid) + ' unknown to oracle.')
            
    def __contains__(self, ghid):
        ''' Checks for the ghid in cache (but does not check for global
        availability; that would require checking the persister for its
        existence and the privateer for access).
        '''
        return ghid in self._lookup class Salmonator(loopa.TaskLooper, metaclass=API):
    ''' Responsible for disseminating Golix objects upstream and
    downstream. Handles all comms with them as well.
    '''
    _golcore = weak_property('__golcore')
    _percore = weak_property('__percore')
    _librarian = weak_property('__librarian')
    _remote_protocol = weak_property('__remote_protocol')
    
    @public_api
    def __init__(self, *args, **kwargs):
        ''' Yarp.
        '''
        super().__init__(*args, **kwargs)
        
        self._deferred = ListMap()
        self._clear_q = None
        
        self._upstream_remotes = set()
        self._downstream_remotes = set()
        
        # Lookup for <registered ghid>
        self._registered = set()
        
        self._bootstrapped = False
        
    @fixture_api
    def RESET(self):
        ''' Clear stuff out, yo.
        '''
        self._upstream_remotes.clear()
        self._downstream_remotes.clear()
        self._registered.clear()
        
    def assemble(self, golcore, percore, librarian, remote_protocol):
        self._golcore = golcore
        self._percore = percore
        self._librarian = librarian
        self._remote_protocol = remote_protocol
    
    @fixture_noop
    @public_api
    def add_upstream_remote(self, task_commander, connection_cls, *args,
                            **kwargs):
        ''' Adds an upstream remote persister.
        *args and **kwargs will be passed to the task_commander task.
        '''
        remote = ConnectionManager(
            connection_cls = connection_cls,
            msg_handler = self._remote_protocol,
            conn_init = self.restore_connection,
        )
        # We have to insert the remote before us at the commander, or shutdown
        # will kill the connections before we can clean them up.
        task_commander.register_task(remote, *args, before_task=self, **kwargs)
        self._upstream_remotes.add(remote)
        
    def add_downstream_remote(self, persister):
        ''' Adds a downstream persister.
        
        PersistenceCore will not attempt to keep a consistent state with
        downstream persisters. Instead, it will simply push updates to
        local objects downstream. It will not, however, look to them for
        updates.
        
        Therefore, to create synchronization **between** multiple
        upstream remotes, also add them as downstream remotes.
        '''
        raise NotImplementedError()
        self._downstream_remotes.add(persister)
        
    async def loop_init(self, *args, **kwargs):
        ''' On top of the usual stuff, set up our queues.
        '''
        await super().loop_init(*args, **kwargs)
        # This "must" be unbounded, because we don't have any control over
        # finalization of objects.
        self._clear_q = asyncio.Queue()
        
    async def loop_stop(self, *args, **kwargs):
        ''' On top of the usual stuff, clear our queues.
        '''
        # Would be good to do, but not currently working
        # for remote in self._persisters:
        #     await self._stop_persister(remote)
        
        await super().loop_stop(*args, **kwargs)
        self._clear_q = None
        
        disconnections = set()
        for remote in self._upstream_remotes:
            if remote.has_connection:
                disconnections.add(
                    make_background_future(remote.disconnect())
                )
        
        # Need to make sure it's not empty
        if disconnections:
            await asyncio.wait(
                fs = disconnections,
                return_when = asyncio.ALL_COMPLETED
            )
        
    async def loop_run(self, *args, **kwargs):
        ''' Wait for object finalizers, and then immediately unsub the
        remotes when the objects are removed from memory.
        '''
        to_clear = await self._clear_q.get()
        await self.deregister(to_clear)
        
    @fixture_noop
    @public_api
    async def notify(self, ghid):
        ''' Notify the salmonator that an upstream subscription
        notification was just successfully completed.
        '''
        # If that was a dynamic object, we also need to pull its target.
        obj = await self._librarian.summarize(ghid)
        if isinstance(obj, _GobdLite):
            if not (await self._librarian.contains(obj.target)):
                await self.pull(obj.target)
    
    @fixture_noop
    @public_api
    async def register(self, ghid):
        ''' Tells the Salmonator to listen upstream for any updates.
        Consumers must independently register an object finalizer for
        salmonator._deregister (not to be called atexit) to remove the
        subscription when the object leaves local memory. TODO: fix that
        leaky abstraction.
        '''
        obj = await self._librarian.summarize(ghid)
        
        if isinstance(obj, _GobdLite):
            logger.info(
                'GAO ' + str(obj.ghid) + ' upstream registration starting.'
            )
            self._registered.add(obj.ghid)
    
            subscriptions = set()
            for remote in self._upstream_remotes:
                if remote.has_connection:
                    subscriptions.add(
                        make_background_future(remote.subscribe(obj.ghid))
                    )
                
            # Need to make sure it's not empty
            if subscriptions:
                await asyncio.wait(
                    fs = subscriptions,
                    return_when = asyncio.ALL_COMPLETED
                )
        
        else:
            logger.debug(
                'GAO ' + str(obj.ghid) + ' cannot be registered upstream: ' +
                'invalid object type: ' + str(type(obj))
            )
    
    @fixture_noop
    @public_api
    def _deregister(self, ghid):
        ''' Finalizer for GAO objects that executes async deregister()
        from within the event loop. Must be called from within our own
        thread, which it should be (finalizers are called from object
        thread, and all gao must be created within event loop's thread).
        '''
        logger.debug('_deregister finalizer called for ' + str(ghid))
        if self._clear_q is not None:
            # This needs to be a function, not a coro, so use nowait.
            self._clear_q.put_nowait(ghid)
    
    @fixture_noop
    @public_api
    async def deregister(self, ghid):
        ''' Tells the salmonator to stop listening for upstream object
        updates.
        '''
        logger.debug('Deregistering updates for ' + str(ghid))
        # This should maybe use remove instead of discard?
        self._registered.discard(ghid)
    
        unsubscriptions = set()
        for remote in self._upstream_remotes:
            unsubscriptions.add(
                make_background_future(remote.unsubscribe(ghid))
            )
        
        # Need to make sure it's not empty
        if unsubscriptions:
            await asyncio.wait(
                fs = unsubscriptions,
                return_when = asyncio.ALL_COMPLETED
            )
    
    @fixture_noop
    @public_api
    async def push(self, ghid):
        ''' Push a single ghid to all remotes.
        
        TODO: smartly defer any pushes to unavailable remotes using
        "if remote.has_connection()"
        '''
        data = await self._librarian.retrieve(ghid)
        
        tasks = set()
        for remote in self._upstream_remotes:
            if remote.has_connection:
                tasks.add(
                    make_background_future(remote.publish(data))
                )
            else:
                self._deferred.append(remote, data)
            
        # Need to make sure it's not empty
        if tasks:
            await asyncio.wait(
                fs = tasks,
                return_when = asyncio.ALL_COMPLETED
            )
    
    @fixture_noop
    @public_api
    async def pull(self, ghid):
        ''' Gets a ghid from upstream. Returns on the first result. Note
        that this is not meant to be called on a dynamic address, as a
        subs update from a slower remote would always be overridden by
        the faster one.
        '''
        pull_complete = None
        tasks_available = set()
        task_to_remote = {}
        for remote in self._upstream_remotes:
            if remote.has_connection:
                task = asyncio.ensure_future(
                    self._attempt_pull_single(ghid, remote)
                )
                tasks_available.add(task)
                task_to_remote[task] = remote
            
        # Wait until the first successful task completion
        # Note that this also shields us against having no tasks
        while tasks_available and not pull_complete:
            finished, pending = await asyncio.wait(
                fs = tasks_available,
                return_when = asyncio.FIRST_COMPLETED
            )
        
            # Despite FIRST_COMPLETED, asyncio may return more than one task
            for task in finished:
                # The task finished, so discard it from the available and reap
                # any exception
                tasks_available.discard(task)
                exc = task.exception()
                
                # If there's been an exception, continue waiting for the rest.
                # Downgrade this to info, since things may be missing from
                # persisters all the time. Let the parent log if, for example,
                # it's missing everywhere.
                if exc is not None:
                    remote = task_to_remote[task]
                    logger.info(
                        'Error while pulling from remote at ' +
                        remote._conn_desc + ':\n' +
                        ''.join(traceback.format_tb(exc.__traceback__)) +
                        repr(exc)
                    )
                    pull_complete = None
                
                # Completed successfully, but it could be a 404 (or other
                # error), which would present as result() = False.
                # Instead of letting the while loop handle this, since more
                # than one task can complete simultaneously, make sure we don't
                # already have a finalized result before blindly assigning the
                # task's result.
                elif not pull_complete:
                    pull_complete = task.result()
                    
                # Multiple tasks completed at once. An earlier one was
                # successful. We need to grab the result to suppress asyncio
                # complaints.
                else:
                    task.result()
                
        # No dice. Either finished is None (no remotes), None (no successful
        # pulls), or False (exactly one remote had the object, but it was
        # unloadable). Raise.
        if not pull_complete:
            raise UnavailableUpstream(
                'Object was unavailable or unacceptable at all '
                'currently-registered remotes.'
            )
            
        # Log success.
        else:
            logger.debug(
                'Successful remote pull for {!s}. Handling...'.format(ghid)
            )
        
        # We may still have some pending tasks. Cancel them. Note that we have
        # not yielded control to the event loop, so there is no race.
        for task in pending:
            logger.debug('Cancelling pending pulls.')
            task.cancel()
            
        # Now handle the result.
        await self._handle_successful_pull(pull_complete)
            
    async def _handle_successful_pull(self, maybe_obj):
        ''' Dispatches the object in the successful pull.
        
        maybe_obj can be either True (if the object already existed
        locally), or the object itself (if the object was new).
        '''
        # If we got a gobd, make sure its target is in the librarian
        if isinstance(maybe_obj, _GobdLite):
            if not (await self._librarian.contains(maybe_obj.target)):
                # Catch unavailableupstream and log a warning.
                # TODO: add logic to retry a few times and then discard
                try:
                    await self.pull(maybe_obj.target)
                    
                except UnavailableUpstream:
                    logger.warning(
                        'Received a subscription notification for ' +
                        str(maybe_obj.ghid) + ', but the sub\'s target was '
                        'missing both locally and upstream.'
                    )
        
        logger.debug('Successful pull handled.')
    
    @fixture_noop
    @public_api
    async def attempt_pull(self, ghid, quiet=False):
        ''' Grabs the ghid from remotes, if available, and puts it into
        the ingestion pipeline.
        '''
        # TODO: check locally, run _inspect, check if mutable before blindly
        # pulling.
        try:
            await self.pull(ghid)
            
        except UnavailableUpstream:
            if not quiet:
                raise
            # Suppress errors if we were called quietly.
            else:
                logger.info(
                    'Object was unavailable or unacceptable upstream, but '
                    'pull was called quietly: ' + str(ghid)
                )
        
    async def _attempt_pull_single(self, ghid, remote):
        ''' Attempt to fetch a single object from a single remote. If
        successful, put it into the ingestion pipeline.
        '''
        # This may error, but any errors here will be caught by the parent.
        data = await remote.get(ghid)
        
        # Call as remotable=False to avoid infinite loops.
        ingested = await self._percore.ingest(data, remotable=False)
        
        # Note that ingest can either return None, if we already have
        # the object, or the object itself, if it's new.
        if ingested:
            return ingested
        else:
            return True
            
    async def bootstrap(self, account):
        ''' Bootstrapping publishes our identity upstream.
        
        ONLY do this if we already have a connection available there, or
        we will hang.
        '''
        for remote in self._upstream_remotes:
            if remote.has_connection:
                await remote.publish(account._identity.second_party.packed)
                await remote.subscribe(account._identity.ghid)
            
        self._bootstrapped = True
    
    @fixture_noop
    @public_api
    async def restore_connection(self, remote, connection):
        ''' Start or re-start a connection.
        '''
        # We need to subscribe to our identity if we're restoring a terminated
        # connection. If we haven't bootstrapped though, that will be handled
        # there.
        if self._bootstrapped:
            await remote.publish(self._golcore._identity.second_party.packed)
            await self._remote_protocol.subscribe(
                connection,
                self._golcore.whoami
            )
        
        # For every every active (salmonator-registered) GAO's ghid...
        tasks = set()
        for registrant in self._registered:
            # Record that we need to perform...
            tasks.add(
                # ...as a background future (which handles its own errors)...
                make_background_future(
                    # ...a subscription call with our connection at protocol
                    self._remote_protocol.subscribe(connection, registrant)
                )
            )
        
        # We need to make sure there's at least one task.
        if tasks:
            # And now just wait for all of that to complete.
            await asyncio.wait(
                fs = tasks,
                return_when = asyncio.ALL_COMPLETED
            )
        
        # Now, we need to destructively iterate over our deferreds until the
        # remote list is exhausted. We have to do this in order, sequentially,
        # and serially, because eg. containers require bindings, etc.
        to_push = collections.deque(self._deferred.pop_key(remote))
        try:
            while to_push:
                data = to_push.popleft()
                await self._remote_protocol.publish(connection, data)
        
        # Restore the last to_push, put it back into the deferred list, and
        # then re-raise
        except Exception:
            to_push.appendleft(data)
            self._deferred.extend(remote, to_push)
            raise class RemotePersistenceProtocol(metaclass=RequestResponseAPI,
                                error_codes=ERROR_CODES,
                                default_version=b'\x00\x00'):
    ''' Defines the protocol for remote persisters.
    '''
    _percore = weak_property('__percore')
    _librarian = weak_property('__librarian')
    _postman = weak_property('__postman')
    _salmonator = weak_property('__salmonator')
    
    @fixture_api
    def __init__(self, percore=None, librarian=None, *args, **kwargs):
        super(RemotePersistenceProtocol.__fixture__, self).__init__(
            *args,
            **kwargs
        )
        if percore is not None:
            self._percore = percore
        if librarian is not None:
            self._librarian = librarian
        
    def assemble(self, percore, librarian, postman, salmonator=None):
        # Link to the remote core.
        self._percore = percore
        self._postman = postman
        self._librarian = librarian
        
        if salmonator is not None:
            self._salmonator = salmonator
        
    @request(b'??')
    async def ping(self, connection):
        ''' Check a remote for availability.
        '''
        return b''
        
    @ping.request_handler
    async def ping(self, connection, body):
        # Really not much to see here.
        return b'\x01'
        
    @ping.response_handler
    async def ping(self, connection, response, exc):
        # This will suppress any errors during pinging.
        if response == b'\x01':
            return True
        else:
            return False
    
    @public_api
    @request(b'PB')
    async def publish(self, connection, packed):
        ''' Publish a packed Golix object.
        '''
        return packed
        
    @publish.fixture
    async def publish(self, connection, packed):
        ''' Just slap the thing into librarian with no checking for
        fixtures.
        '''
        obj = await self._percore.attempt_load(packed)
        await self._librarian.store(obj, packed)
        
    @publish.request_handler
    async def publish(self, connection, body):
        ''' Handle a published object.
        '''
        obj = await self._percore.ingest(
            packed = body,
            remotable = False,
            skip_conn = weakref.ref(connection)
        )
        
        # Object already existed.
        if obj is None:
            response = b'\x00'
        
        # Object is new
        else:
            response = b'\x01'
            
        # We don't need to wait for the mail run to have a successful return
        return response
        
    @publish.response_handler
    async def publish(self, connection, response, exc):
        ''' Handle responses to publish requests.
        '''
        if exc is not None:
            raise exc
        else:
            return True
    
    @public_api
    @request(b'GT')
    async def get(self, connection, ghid):
        ''' Request an object from the persistence provider.
        '''
        return bytes(ghid)
        
    @get.fixture
    async def get(self, connection, ghid):
        ''' Fixture to just pull directly from librarian.
        '''
        return (await self._librarian.retrieve(ghid))
        
    @get.request_handler
    async def get(self, connection, body):
        ''' Handle get requests.
        '''
        ghid = Ghid.from_bytes(body)
        return (await self._librarian.retrieve(ghid))
    
    @public_api
    @request(b'+S')
    async def subscribe(self, connection, ghid):
        ''' Subscribe to updates from the remote.
        '''
        return bytes(ghid)
        
    @subscribe.fixture
    async def subscribe(self, connection, ghid):
        ''' Manual noop.
        '''
        
    @subscribe.request_handler
    async def subscribe(self, connection, body):
        ''' Handle subscription requests.
        '''
        ghid = Ghid.from_bytes(body)
        await self._postman.subscribe(connection, ghid)
        return b'\x01'
        
    @request(b'-S')
    async def unsubscribe(self, connection, ghid):
        ''' Unsubscribe from updates at a remote.
        '''
        return bytes(ghid)
        
    @unsubscribe.request_handler
    async def unsubscribe(self, connection, body):
        ''' Handle unsubscription requests.
        '''
        ghid = Ghid.from_bytes(body)
        had_subscription = await self._postman.unsubscribe(connection, ghid)
        
        if had_subscription:
            return b'\x01'
        
        # Still successful, but idempotent
        else:
            return b'\x00'
        
    @unsubscribe.response_handler
    async def unsubscribe(self, connection, response, exc):
        ''' Handle responses to unsubscription requests.
        '''
        if exc is not None:
            raise exc
            
        # For now, ignore (success & UNsub) vs (success & NOsub)
        else:
            return True
    
    @public_api
    @request(b'!!')
    async def subscription_update(self, connection, subscription_ghid,
                                  notification_ghid):
        ''' Send a subscription update to the connection.
        '''
        payload = await self._librarian.retrieve(notification_ghid)
        return bytes(subscription_ghid) + payload
        
    @subscription_update.fixture
    async def subscription_update(self, connection, subscription_ghid,
                                  notification_ghid, timeout=None):
        ''' Make a manual no-op fixture, since inspect signatures
        apparently don't from_callable on a descriptor... (grrr). Also,
        because timeout isn't appropriately wrapped.
        '''
        
    @subscription_update.request_handler
    async def subscription_update(self, connection, body):
        ''' Handles an incoming subscription update.
        '''
        subscribed_ghid = Ghid.from_bytes(body[0:65])
        notification = body[65:]
        
        try:
            # Note that this handles postman scheduling as well.
            ingested = await self._percore.ingest(
                notification,
                remotable = False,
                skip_conn = connection
            )
            
        except AlreadyDebound as exc:
            vomitus = exc.ghid
            debindings = await self._librarian.debind_status(vomitus)
            debinding_ghid = next(debinding for debinding in debindings)
            logger.info(str(subscribed_ghid) +
                        ' subscription update already debound: ' +
                        str(vomitus) + '. Pushing debinding upstream: ' +
                        str(debinding_ghid))
            
            make_background_future(
                self._salmonator.push(debinding_ghid)
            )
            
        else:
            # But, if ingested, we need to notify the salmonator, so it can (if
            # needed) also acquire the target
            if ingested:
                make_background_future(
                    self._salmonator.notify(subscribed_ghid)
                )
        
        return b'\x01'
        
    @request(b'?S')
    async def query_subscriptions(self, connection):
        ''' Request a list of all currently subscribed ghids.
        '''
        return b''
        
    @query_subscriptions.request_handler
    async def query_subscriptions(self, connection, body):
        ''' Handle subscription query requests.
        '''
        ghidlist = list(await self._postman.list_subs(connection))
        parser = generate_ghidlist_parser()
        return parser.pack(ghidlist)
        
    @query_subscriptions.response_handler
    async def query_subscriptions(self, connection, response, exc):
        ''' Handle responses to subscription queries.
        '''
        if exc is not None:
            raise exc
        
        parser = generate_ghidlist_parser()
        return set(parser.unpack(response))
        
    @request(b'?B')
    async def query_bindings(self, connection, ghid):
        ''' Get a list of all bindings for the ghid.
        '''
        return bytes(ghid)
        
    @query_bindings.request_handler
    async def query_bindings(self, connection, body):
        ''' Handle binding query requests.
        '''
        ghid = Ghid.from_bytes(body)
        ghidlist = await self._librarian.bind_status(ghid)
        parser = generate_ghidlist_parser()
        return parser.pack(list(ghidlist))
        
    @query_bindings.response_handler
    async def query_bindings(self, connection, response, exc):
        ''' Handle responses to binding queries.
        '''
        if exc is not None:
            raise exc
            
        parser = generate_ghidlist_parser()
        return set(parser.unpack(response))
        
    @request(b'?D')
    async def query_debindings(self, connection, ghid):
        ''' Query which, if any, ghid(s) have debindings for <ghid>.
        '''
        return bytes(ghid)
        
    @query_debindings.request_handler
    async def query_debindings(self, connection, body):
        ''' Handles debinding query requests.
        '''
        ghid = Ghid.from_bytes(body)
        ghidlist = await self._librarian.debind_status(ghid)
        parser = generate_ghidlist_parser()
        return parser.pack(list(ghidlist))
        
    @query_debindings.response_handler
    async def query_debindings(self, connection, response, exc):
        ''' Handle responses to debinding queries.
        '''
        if exc is not None:
            raise exc
            
        parser = generate_ghidlist_parser()
        return set(parser.unpack(response))
    
    @public_api
    @request(b'?E')
    async def query_existence(self, connection, ghid):
        ''' Checks if the passed <ghid> exists at the remote.
        '''
        return bytes(ghid)
        
    @query_existence.fixture
    async def query_existence(self, connection, ghid):
        ''' Just always return false.
        '''
        return (await self._librarian.contains(ghid))
        
    @query_existence.request_handler
    async def query_existence(self, connection, body):
        ''' Handle existence queries.
        '''
        ghid = Ghid.from_bytes(body)
        if (await self._librarian.contains(ghid)):
            return b'\x01'
        else:
            return b'\x00'
        
    @query_existence.response_handler
    async def query_existence(self, connection, response, exc):
        ''' Handle responses to existence queries.
        '''
        if exc is not None:
            raise exc
        elif response == b'\x00':
            return False
        else:
            return True
        
    @request(b'XX')
    async def disconnect(self, connection):
        ''' Terminates all subscriptions and requests.
        '''
        return b''
        
    @disconnect.request_handler
    async def disconnect(self, connection, body):
        ''' Handle disconnect requests.
        '''
        await self._postman.clear_subs(connection)
        return b'\x01' class _Dispatchable(GAOCore, metaclass=API):
    ''' A dispatchable object. Note that privacy is store within the
    dispatcher itself, and not within the individual dispatchables.
    '''
    _dispatch = weak_property('__dispatch')
    _account = weak_property('__account')
    _ipc_protocol = weak_property('__ipc_protocol')
    api_id = immutable_property('_api_id')
    
    def __init__(self, *args, api_id, state, dispatch, ipc_protocol, account,
                 **kwargs):
        super().__init__(*args, **kwargs)
        
        self._dispatch = dispatch
        self._ipc_protocol = ipc_protocol
        self._account = account
        
        # These may be explicitly set to None to allow pull() to apply them.
        if state is not None:
            self.state = state
            
        if api_id is not None:
            self.api_id = api_id
            
    @public_api
    async def _push(self):
        ''' Extend GAOCore._push() to flush the account upon completion.
        '''
        await super()._push()
        await self._account.flush()
        
    @_push.fixture
    async def _push(self):
        ''' Call super() to FIXTURE super instead of dispatchable.
        '''
        await super(_Dispatchable, self)._push()
    
    @public_api
    async def _pull(self):
        ''' Extend GAOCore._pull() to notify IPC about updates.
        '''
        await super()._pull()
        make_background_future(
            self._dispatch.distribute_update(self.ghid)
        )
        
    @_pull.fixture
    async def _pull(self):
        ''' Call super() to FIXTURE super instead of dispatchable.
        '''
        await super(_Dispatchable, self)._pull()
        
    async def pack_gao(self):
        ''' Packs state into a bytes object. May be overwritten in subs
        to pack more complex objects. Should always be a staticmethod or
        classmethod.
        '''
        version = b'\x00'
        return b'hgxd' + version + bytes(self.api_id) + self.state
        
    async def unpack_gao(self, packed):
        ''' Unpacks state from a bytes object. May be overwritten in
        subs to unpack more complex objects. Should always be a
        staticmethod or classmethod.
        '''
        magic = packed[0:4]
        version = packed[4:5]
        api_id = ApiID.from_bytes(packed[5:70])
        
        if magic != b'hgxd':
            raise DispatchError('Object does not appear to be dispatchable.')
        elif version != b'\x00':
            raise DispatchError('Incompatible dispatchable version number.')
        elif getattr(self, 'api_id', None) is None:
            self.api_id = api_id
        else:
            # YES, this does need to be _api_id (with underscore!) because
            # we may need to override something existing.
            self._api_id = api_id
        # Checking for this from other people isn't really that useful. We'll
        # already error out if you try to change it locally, and if someone
        # goes to the effort of changing it elsewhere, well, more power to them
        # I guess.
        # elif api_id != self.api_id:
        #     raise DispatchError('Cannot change API ID.')
            
        self.state = packed[70:]
    
    @fixture_noop
    @public_api
    async def apply_delete(self, debinding):
        ''' Extend GAOCore.apply_delete to notify consumers of deletion.
        '''
        await super().apply_delete(debinding)
        make_background_future(
            self._dispatch.distribute_update(self.ghid, deleted=True)
        )
        
    def __eq__(self, other):
        ''' Ensure the other object is the same gao with the same state.
        Intended pretty much exclusively for use in testing.
        '''
        equal = True
        
        try:
            equal &= (self.dynamic == other.dynamic)
            equal &= (self.ghid == other.ghid)
            equal &= (self.author == other.author)
            equal &= (self.state == other.state)
            equal &= (self.api_id == other.api_id)
            
        except AttributeError:
            equal = False
        
        return equal
        
    def __hash__(self):
        ''' Dispatchable hashes should mix the hashes of the api_id, the
        author, and the ghid together, but only if all of the above are
        defined.
        '''
        defined = self.ghid is not None
        defined &= self.author is not None
        defined &= self.api_id is not None
        
        if not defined:
            raise TypeError(
                'Dispatchable objects may only be hashed after they are ' +
                'fully defined (with a ghid, an author, and an api_id.'
            )
        
        return hash(self.ghid) ^ hash(self.author) ^ hash(self.api_id) class Enforcer(metaclass=API):
    ''' Enforces valid target selections.
    '''
    _librarian = weak_property('__librarian')
    
    @fixture_api
    def __init__(self, librarian, *args, **kwargs):
        super(Enforcer.__fixture__, self).__init__(*args, **kwargs)
        self._librarian = librarian
        
    def assemble(self, librarian):
        # Call before using.
        self._librarian = librarian
        
    async def validate_gidc(self, obj):
        ''' GIDC need no target verification.
        '''
        return True
        
    async def validate_geoc(self, obj):
        ''' GEOC need no target validation.
        '''
        return True
        
    async def validate_gobs(self, obj):
        ''' Check if target is known, and if it is, validate it.
        '''
        try:
            target = await self._librarian.summarize(obj.target)
        # TODO: think more about this, and whether everything has been updated
        # appropriately to raise a DoesNotExist instead of a KeyError.
        # This could be more specific and say DoesNotExist
        except KeyError:
            logger.debug(str(obj) + ' target missing from librarian: ' +
                         str(obj.target))
        else:
            for forbidden in (_GidcLite, _GobsLite, _GdxxLite, _GarqLite):
                if isinstance(target, forbidden):
                    raise InvalidTarget(str(obj) + ' target invalid: ' +
                                        str(target))
        return True
        
    async def validate_gobd(self, obj):
        ''' Check if target is known, and if it is, validate it.
        
        Also do a state check on the dynamic binding.
        '''
        try:
            target = await self._librarian.summarize(obj.target)
        except KeyError:
            logger.debug(str(obj) + ' target missing from librarian: ' +
                         str(obj.target))
        else:
            for forbidden in (_GidcLite, _GobsLite, _GdxxLite, _GarqLite):
                if isinstance(target, forbidden):
                    raise InvalidTarget(str(obj) + ' target invalid: ' +
                                        str(target))
                    
        await self._validate_dynamic_history(obj)
                    
        return True
        
    async def validate_gdxx(self, obj, target_obj=None):
        ''' Check if target is known, and if it is, validate it.
        '''
        try:
            if target_obj is None:
                target = await self._librarian.summarize(obj.target)
            else:
                target = target_obj
        except KeyError:
            logger.warning(str(obj) + ' validated by Enforcer, but its ' +
                           'target was unknown: ' + str(obj.target))
            logger.debug(str(obj) + ' missing target traceback:\n' +
                         ''.join(traceback.format_exc()))
            
        else:
            # NOTE: if this changes, will need to modify place_gdxx in _Bookie
            for forbidden in (_GidcLite, _GeocLite):
                if isinstance(target, forbidden):
                    raise InvalidTarget(str(obj) + ' target invalid: ' +
                                        str(target))
        return True
        
    async def validate_garq(self, obj):
        ''' No additional validation needed.
        '''
        return True
        
    async def _validate_dynamic_history(self, obj):
        ''' Enforces state flow / progression for dynamic objects. In
        other words, ensures monotonic counter.
        '''
        # Try getting an existing binding.
        try:
            existing = await self._librarian.summarize(obj.ghid)
        
        # TOFU (trust on first upload lulz)
        except KeyError:
            pass
                
        else:
            if existing.counter >= obj.counter:
                logger.debug('New obj frame:     ' +
                             str(obj.frame_ghid))
                logger.debug('New obj tarvec:    ' +
                             str(obj.target_vector))
                logger.debug('Existing frame:    ' +
                             str(existing.frame_ghid))
                logger.debug('Existing tarvec:   ' +
                             str(existing.target_vector))
                raise IllegalDynamicFrame(str(obj.ghid) + ' counter ' +
                                          'decreased from ' +
                                          str(existing.counter) + ' to ' +
                                          str(obj.counter)) class PersistenceCore(metaclass=API):
    ''' Provides the core functions for storing Golix objects. Required
    for the hypergolix service to start.
    
    Can coordinate with both "upstream" and "downstream" persisters.
    Other persisters should pass through the "ingestive tract". Local
    objects can be published directly through calling the ingest_<type>
    methods.
    
    TODO: add librarian validation, so that attempting to update an
    object we already have an identical copy to silently exits.
    '''
    _doorman = weak_property('__doorman')
    _enforcer = weak_property('__enforcer')
    _lawyer = weak_property('__lawyer')
    _bookie = weak_property('__bookie')
    _postman = weak_property('__postman')
    _undertaker = weak_property('__undertaker')
    _librarian = weak_property('__librarian')
    _salmonator = weak_property('__salmonator')
    
    # This is a messy way of getting the suffix for validation and stuff but
    # it's getting the job done.
    _ATTR_LOOKUP = {
        _GidcLite: 'gidc',
        _GeocLite: 'geoc',
        _GobsLite: 'gobs',
        _GobdLite: 'gobd',
        _GdxxLite: 'gdxx',
        _GarqLite: 'garq'
    }
    
    @public_api
    def __init__(self, loop=None, *args, **kwargs):
        ''' Create a KeyedAsyncioLock for ingestion.
        '''
        self._ingestion_mutex = KeyedAsyncioLock(loop)
        
    @__init__.fixture
    def __init__(self, librarian=None, *args, **kwargs):
        ''' Right now, well, this is basically a noop. Anticipating
        substantial changes!
        '''
        super(PersistenceCore.__fixture__, self).__init__(
            *args,
            **kwargs
        )
        
        if librarian is not None:
            self._librarian = librarian
        
    def assemble(self, doorman, enforcer, lawyer, bookie, librarian, postman,
                 undertaker, salmonator):
        self._doorman = doorman
        self._enforcer = enforcer
        self._lawyer = lawyer
        self._bookie = bookie
        self._postman = postman
        self._undertaker = undertaker
        self._librarian = librarian
        self._salmonator = salmonator
    
    @public_api
    async def direct_ingest(self, obj, packed, remotable, skip_conn=None):
        ''' Standard ingestion flow for stuff. To be called from ingest
        above, or directly (for objects created "in-house").
        '''
        # Check for a redundant object, which will immediately short-circuit.
        if isinstance(obj, _GobdLite):
            check_ghid = obj.frame_ghid
            log_frame = True
            counter = obj.counter
            target = obj.target
        else:
            check_ghid = obj.ghid
            log_frame = False
            counter = 0
            target = None
        
        async with self._ingestion_mutex(check_ghid):
            if (await self._librarian.contains(check_ghid)):
                logger.debug(
                    str(check_ghid) + ' not ingested: already exists.'
                )
                return False
            
            else:
                logger.info(
                    'Ingesting ' + str(obj) +
                    (
                        ' frame ' + str(counter) + ': ' + str(check_ghid) +
                        ', target: ' + str(target)
                    ) * log_frame + '...'
                )
                # Calculate "gidc", etc
                suffix = self._ATTR_LOOKUP[type(obj)]
                validation_method = 'validate_' + suffix
                
                # Validate the object... (will raise for invalid)
                # ########################
                # Enforce target selection
                await getattr(self._enforcer, validation_method)(obj)
                # Now make sure authorship requirements are satisfied
                await getattr(self._lawyer, validation_method)(obj)
                # Finally make sure persistence rules are followed
                await getattr(self._bookie, validation_method)(obj)
                
                # Ingest the object
                # ########################
                # Alert the undertaker for any necessary GC of targets, etc. Do
                # that before storing at the librarian, so that the undertaker
                # has access to the old state.
                await getattr(self._undertaker, 'alert_' + suffix)(obj,
                                                                   skip_conn)
                # Finally, add it to the librarian.
                await self._librarian.store(obj, packed)
                
                if remotable:
                    await self._salmonator.push(obj.ghid)
            
                return True
    
    @direct_ingest.fixture
    async def direct_ingest(self, obj, packed, remotable, skip_conn=None):
        ''' Fixture direct ingest to just store at librarian directly
        with no verification (if librarian exists), otherwise simply
        return None.
        '''
        try:
            librarian = self._librarian
        except AttributeError:
            return None
        else:
            await librarian.store(obj, packed)
    
    @public_api
    async def attempt_load(self, packed, quiet=True):
        ''' Attempt to load a packed golix object. If quiet=False, it
        will raise if there is no loader.
        
        TODO: move into doorman.
        '''
        # This is kinda silly, but instead of spewing off a million different
        # threads that will immediately die, let's pre-select the loader
        magic = bytes(packed[:4])
        loaders = {
            b'GIDC': self._doorman.load_gidc,
            b'GEOC': self._doorman.load_geoc,
            b'GOBS': self._doorman.load_gobs,
            b'GOBD': self._doorman.load_gobd,
            b'GDXX': self._doorman.load_gdxx,
            b'GARQ': self._doorman.load_garq
        }
        
        try:
            loader = loaders[magic]
            
        # If no successful loader was found, return None, and allow the parent
        # to raise.
        except KeyError as exc:
            if quiet:
                return None
            else:
                raise MalformedGolixPrimitive('No loader found for magic: ' +
                                              str(magic)) from exc
        
        obj = await loader(packed)
            
        return obj
        
    @attempt_load.fixture
    async def attempt_load(self, packed):
        ''' Create an ad-hoc doorman fixture.
        '''
        # Note that, because of the weak ref, we need to actually hold this
        # here to prevent the doorman from being gc'd. So, just create a
        # disposable one.
        doorman = Doorman.__fixture__()
        self._doorman = doorman
        result = \
            await super(PersistenceCore.__fixture__, self).attempt_load(packed)
        del doorman
        return result
    
    @fixture_noop
    @public_api
    async def ingest(self, packed, remotable=True, skip_conn=None):
        ''' Called on an untrusted and unknown object. May be bypassed
        by locally-created, trusted objects (by calling the individual
        ingest methods directly). Parses, validates, and stores the
        object, and returns True; or, raises an error.
        '''
        # This may return None, but that will be caught by the KeyError below.
        obj = await self.attempt_load(packed)
        if obj is None:
            raise MalformedGolixPrimitive(
                'Packed bytes do not appear to be a Golix primitive.'
            )
        
        ingested = await self.direct_ingest(obj, packed, remotable)
        
        # Note that individual ingest methods are only called directly for
        # locally-built objects, which do not need a mail run.
        # If the object is identical to what we already have, the ingester
        # will return None.
        if ingested:
            await self._postman.schedule(obj, skip_conn=skip_conn)
            
        else:
            logger.debug(str(obj.ghid) + ' postal scheduling aborted on ' +
                         'unchanged object.')
        
        # Note: this is not the place for salmonator pushing! Locally
        # created/updated objects call the individual ingest methods
        # directly, so they have to be the ones that actually deal with it.
        
        return ingested class _GidcLite(_BaseLite):
    ''' Lightweight description of a GIDC.
    '''
    __slots__ = [
        'identity'
    ]
    
    def __init__(self, ghid, identity):
        self.ghid = ghid
        self.identity = identity
        
    @classmethod
    def from_golix(cls, golix_obj):
        ''' Convert the golix object to a lightweight representation.
        '''
        identity = SecondParty.from_identity(golix_obj)
        return cls(
            ghid = golix_obj.ghid,
            identity = identity,
        ) class _GeocLite(_BaseLite):
    ''' Lightweight description of a GEOC.
    '''
    __slots__ = [
        'author',
    ]
    
    def __init__(self, ghid, author):
        self.ghid = ghid
        self.author = author
        
    def __eq__(self, other):
        try:
            return (
                super().__eq__(other) and
                self.author == other.author
            )
        # This will not catch a super() TyperError, so we want to be able to
        # compare anything with a ghid. In reality, any situation where the
        # authors don't match but the ghids do is almost certainly a bug; but,
        # compare it anyways just in case.
        except AttributeError as exc:
            return False
        
    @classmethod
    def from_golix(cls, golix_obj):
        ''' Convert the golix object to a lightweight representation.
        '''
        return cls(
            ghid = golix_obj.ghid,
            author = golix_obj.author,
        ) class _GobsLite(_BaseLite):
    ''' Lightweight description of a GOBS.
    '''
    __slots__ = [
        'author',
        'target',
    ]
    
    def __init__(self, ghid, author, target):
        self.ghid = ghid
        self.author = author
        self.target = target
        
    def __eq__(self, other):
        try:
            return (
                super().__eq__(other) and
                self.author == other.author and
                self.target == other.target
            )
            
        # This will not catch a super() TyperError, so we want to be able to
        # compare anything with a ghid. In reality, any situation where the
        # authors don't match but the ghids do is almost certainly a bug; but,
        # compare it anyways just in case.
        except AttributeError as exc:
            return False
        
    @classmethod
    def from_golix(cls, golix_obj):
        ''' Convert the golix object to a lightweight representation.
        '''
        return cls(
            ghid = golix_obj.ghid,
            author = golix_obj.binder,
            target = golix_obj.target,
        ) class _GobdLite(_BaseLite):
    ''' Lightweight description of a GOBD.
    '''
    __slots__ = [
        'author',
        'counter',
        'target_vector',
        'frame_ghid'
    ]
    
    def __init__(self, ghid, author, counter, target_vector, frame_ghid):
        self.ghid = ghid
        self.author = author
        self.counter = counter
        self.target_vector = target_vector
        self.frame_ghid = frame_ghid
        
    def __eq__(self, other):
        try:
            return (
                super().__eq__(other) and
                self.author == other.author and
                self.target == other.target and
                self.frame_ghid == other.frame_ghid
                # Skip history, because it could potentially vary
                # self.history == other.history
            )
            
        # This will not catch a super() TyperError, so we want to be able to
        # compare anything with a ghid. In reality, any situation where the
        # authors don't match but the ghids do is almost certainly a bug; but,
        # compare it anyways just in case.
        except AttributeError as exc:
            return False
        
    @classmethod
    def from_golix(cls, golix_obj):
        ''' Convert the golix object to a lightweight representation.
        '''
        return cls(
            ghid = golix_obj.ghid_dynamic,
            author = golix_obj.binder,
            counter = golix_obj.counter,
            target_vector = golix_obj.target_vector,
            frame_ghid = golix_obj.ghid
        )
        
    @property
    def target(self):
        ''' Access the current target.
        '''
        return self.target_vector[0] class _GdxxLite(_BaseLite):
    ''' Lightweight description of a GDXX.
    '''
    __slots__ = [
        'author',
        'target',
        '_debinding',
    ]
    
    def __init__(self, ghid, author, target):
        self.ghid = ghid
        self.author = author
        self.target = target
        self._debinding = True
        
    def __eq__(self, other):
        try:
            return (
                super().__eq__(other) and
                self.author == other.author and
                self._debinding == other._debinding
            )
            
        # This will not catch a super() TyperError, so we want to be able to
        # compare anything with a ghid. In reality, any situation where the
        # authors don't match but the ghids do is almost certainly a bug; but,
        # compare it anyways just in case.
        except AttributeError as exc:
            return False
        
    @classmethod
    def from_golix(cls, golix_obj):
        ''' Convert the golix object to a lightweight representation.
        '''
        return cls(
            ghid = golix_obj.ghid,
            author = golix_obj.debinder,
            target = golix_obj.target,
        ) class _GarqLite(_BaseLite):
    ''' Lightweight description of a GARQ.
    '''
    __slots__ = [
        'recipient',
    ]
    
    def __init__(self, ghid, recipient):
        self.ghid = ghid
        self.recipient = recipient
        
    def __eq__(self, other):
        try:
            return (
                super().__eq__(other) and
                self.recipient == other.recipient
            )
            
        # This will not catch a super() TyperError, so we want to be able to
        # compare anything with a ghid. In reality, any situation where the
        # authors don't match but the ghids do is almost certainly a bug; but,
        # compare it anyways just in case.
        except AttributeError as exc:
            return False
        
    @classmethod
    def from_golix(cls, golix_obj):
        ''' Convert the golix object to a lightweight representation.
        '''
        return cls(
            ghid = golix_obj.ghid,
            recipient = golix_obj.recipient,
        ) class InvalidIdentity(RemoteNak):
    ''' This RemoteNak is raised when a persistence provider has received
    a primitive with an unknown or invalid author or recipient.
    '''
    pass class InconsistentAuthor(RemoteNak):
    ''' This RemoteNak is raised when a persistence provider has received
    a primitive targeting an inappropriate object (for example, trying
    to debind a GEOC directly).
    '''
    pass class ApiID(Ghid):
    ''' Subclass Ghid in a way that makes the API ID seem like a normal
    64-byte string.
    
    Remind me again why the hell I'm subclassing ghid for this when I'm
    ending up removing just about everything that makes it a ghid?
    '''
    
    def __init__(self, address, algo=None):
        ''' Wrap the normal Ghid creation with a forced algo.
        '''
        if len(address) != 64:
            raise ValueError('Improper API ID length.')
        
        elif algo is None:
            algo = 0
            
        elif algo != 0:
            raise ValueError('Improper API ID format.')
        
        super().__init__(algo, address)
        
    def __repr__(self):
        ''' Hide that this is a ghid.
        '''
        return type(self).__name__ + '(' + repr(self.address) + ')'
        
    @property
    def algo(self):
        return self._algo
        
    @algo.setter
    def algo(self, value):
        # Currently, ensure algo is b'\x00'
        if value == 0:
            self._algo = value
        else:
            raise ValueError('Invalid address algorithm.')
            
    @property
    def address(self):
        return self._address
            
    @address.setter
    def address(self, address):
        self._address = address
        
    @classmethod
    def pseudorandom(cls):
        return super().pseudorandom(0)
import unittest
import collections
import queue
import random
import inspect
import asyncio
from loopa import NoopLoop
from loopa.utils import await_coroutine_threadsafe
from hypergolix.librarian import LibrarianCore
from hypergolix.postal import PostalCore
from hypergolix.postal import PostOffice
from hypergolix.postal import MrPostman
from hypergolix.core import GolixCore
from hypergolix.rolodex import Rolodex
from hypergolix.core import Oracle
from hypergolix.remotes import Salmonator
from hypergolix.remotes import RemotePersistenceProtocol
from hypergolix.dispatch import _Dispatchable
from hypergolix.persistence import Enforcer
from hypergolix.persistence import PersistenceCore
from hypergolix.persistence import _GidcLite
from hypergolix.persistence import _GeocLite
from hypergolix.persistence import _GobsLite
from hypergolix.persistence import _GobdLite
from hypergolix.persistence import _GdxxLite
from hypergolix.persistence import _GarqLite
from hypergolix.exceptions import InvalidIdentity
from hypergolix.exceptions import InconsistentAuthor
from hypergolix.utils import ApiID
from golix._getlow import GIDC
from _fixtures.ghidutils import make_random_ghid
from _fixtures.identities import TEST_AGENT1
from _fixtures.identities import TEST_AGENT2
from _fixtures.identities import TEST_READER1
from _fixtures.identities import TEST_READER2
from _fixtures.remote_exchanges import gidc1
from _fixtures.remote_exchanges import gidc2
from _fixtures.remote_exchanges import cont1_1  # Known author
from _fixtures.remote_exchanges import cont3_1  # Unknown author
from _fixtures.remote_exchanges import bind1_1  # Known author
from _fixtures.remote_exchanges import bind3_1  # Unknown author
from _fixtures.remote_exchanges import dyn1_1a  # Known author frame 1
from _fixtures.remote_exchanges import dyn1_1b  # Known author frame 2
from _fixtures.remote_exchanges import dyn3_1a  # Unknown author frame 1
from _fixtures.remote_exchanges import dyn3_1b  # Unknown author frame 2
from _fixtures.remote_exchanges import dynF_a   # Inconsistent author frame 1
from _fixtures.remote_exchanges import dynF_b   # Inconsistent author frame 2
from _fixtures.remote_exchanges import dynF_c   # Inconsistent, unk author frm2
from _fixtures.remote_exchanges import debind1_1        # Consistent author
from _fixtures.remote_exchanges import debind1_F        # Inconsistent author
from _fixtures.remote_exchanges import debind3_1        # Unknown author
from _fixtures.remote_exchanges import debindR_1
from _fixtures.remote_exchanges import debindR_F
from _fixtures.remote_exchanges import debind3_TF
from _fixtures.remote_exchanges import dyndebind1_1
from _fixtures.remote_exchanges import dedebind1_1
from _fixtures.remote_exchanges import handshake1_1     # Known recipient
from _fixtures.remote_exchanges import handshake3_1     # Unknown recipient
    from hypergolix import logutils
        )
        self.assertTrue(len(scheduled) == 1)
        removal = scheduled.pop()
        self.assertEqual(removal, (dbind1a.ghid, xbind1d.ghid, None))
        
        # Test with normal (non-delete), but missing object (should defer)
        await_coroutine_threadsafe(
            coro = self.postman.schedule(dbind1a),
            loop = self.nooploop._loop
        )
        scheduled = await_coroutine_threadsafe(
            coro = self.postman.get_scheduled(),
            loop = self.nooploop._loop
        )
        self.assertTrue(len(scheduled) == 0)
        # Note that this also resets deferred to pristine state!
        deferred = self.postman._deferred.pop_any(obj1.ghid)
        self.assertIn((dbind1a.ghid, dbind1a.frame_ghid, None), deferred)
        
        # Finally, test with normal (non-delete), and existant object
        await_coroutine_threadsafe(
            coro = self.librarian.store(obj1, cont1_1.packed),
            loop = self.nooploop._loop
        )
        await_coroutine_threadsafe(
            coro = self.postman.schedule(dbind1a),
            loop = self.nooploop._loop
        )
        scheduled = await_coroutine_threadsafe(
            coro = self.postman.get_scheduled(),
            loop = self.nooploop._loop
        )
        self.assertTrue(len(scheduled) == 1)
        self.assertEqual(
            scheduled.pop(), (dbind1a.ghid, dbind1a.frame_ghid, None)
        )
        
    def test_gdxx(self):
        ''' Test gdxx operations.
        '''
        await_coroutine_threadsafe(
            coro = self.librarian.store(xbind1, debind1_1.packed),
            loop = self.nooploop._loop
        )
        
        # Gidc should never notify
        await_coroutine_threadsafe(
            coro = self.postman.schedule(xbind1),
            loop = self.nooploop._loop
        )
        
        scheduled = await_coroutine_threadsafe(
            coro = self.postman.get_scheduled(),
            loop = self.nooploop._loop
        )
        self.assertTrue(len(scheduled) == 0)
        
    def test_garq(self):
        ''' Test garq operations.
        '''
        # Prep with dynamic binding and debinding
        await_coroutine_threadsafe(
            coro = self.librarian.store(req1, handshake1_1.packed),
            loop = self.nooploop._loop
        )
        await_coroutine_threadsafe(
            coro = self.librarian.store(xbind1R, debindR_1.packed),
            loop = self.nooploop._loop
        )
        
        # Removal should generate a notification
        await_coroutine_threadsafe(
            coro = self.postman.schedule(req1, removed=True),
            loop = self.nooploop._loop
        )
        scheduled = await_coroutine_threadsafe(
            coro = self.postman.get_scheduled(),
            loop = self.nooploop._loop
        )
        self.assertTrue(len(scheduled) == 1)
        removal = scheduled.pop()
        self.assertEqual(removal, (req1.recipient, xbind1R.ghid, None))
        
        # As should non-removal
        await_coroutine_threadsafe(
            coro = self.postman.schedule(req1),
            loop = self.nooploop._loop
        )
        scheduled = await_coroutine_threadsafe(
            coro = self.postman.get_scheduled(),
            loop = self.nooploop._loop
        )
        self.assertTrue(len(scheduled) == 1)
        notification = scheduled.pop()
        self.assertEqual(notification, (req1.recipient, req1.ghid, None))
        
        
class MrPostmanTest(unittest.TestCase):
    ''' Test postman for local persistence systems.
    '''
    
    @classmethod
    def setUpClass(cls):
        cls.nooploop = NoopLoop(
            debug = True,
            threaded = True
        )
        cls.nooploop.start()
        
    @classmethod
    def tearDownClass(cls):
        # Kill the running loop.
        cls.nooploop.stop_threadsafe_nowait()
    
    def setUp(self):
        ''' Do a per-test fresh init of the undertaker, as well as the
        fixtures for both librarian and postman.
        '''
        # First prep fixtures
        self.librarian = LibrarianCore.__fixture__()

        self.rolodex = Rolodex.__fixture__()