diff --git a/bin/fm-courier-pickup.py b/bin/fm-courier-pickup.py index 38999e629d1..3f06d33a619 100755 --- a/bin/fm-courier-pickup.py +++ b/bin/fm-courier-pickup.py @@ -35,7 +35,8 @@ the vc- note and appends its check wake, which the watcher's conversation ring surfaces within about a second (docs/watcher-continuity.md). A capture that fails is retried after 5 and 20 seconds, then, unless the transport shows it -landed, the captain is told dispatch FAILURE text. An empty transcript is not +landed, the captain is told dispatch FAILURE text. A restart retries the pending +capture immediately, retaining its attempt count. An empty transcript is not filed and he is told NOT_TEXT. With every transport call bounded at 10 s, the failure notice is normally queued within about 75 s of reading the message, and its failed stage follows the notice's sent receipt. If the notice waits @@ -73,7 +74,8 @@ Texts go one at a time in order: the next is published only after the courier's terminal result receipt for the previous one. If a consumed request has no result receipt, pickup republishes its identical -saved request at most once per 60 s to recover the receipt using the same ID. +saved request after 60 s to recover the receipt using the same ID; a last +publication timestamp ahead of wall time makes recovery due immediately. A sent receipt records playback completed and moves the stage the reply implies (receipt/progress working, question question, error failed, final answer done). Stages are published as ID.stage.json in @@ -83,8 +85,12 @@ FM_COURIER_ROOT (default /) relocates srv/courier and etc/courier for offline tests; FM_COURIER_USER (default courier) names the record and receipt owner. The binding is FM_HOME/state/imessage/binding.json, exactly -{conversation_id, credential, destination: "imessage"}. Startup refusals exit 1, -usage 2; per-tick failures are logged and the loop continues. +{conversation_id, credential, destination: "imessage"} as fm-inbox.sh +conversation bind returned it. Startup refusals exit 1, usage 2; per-tick +failures are logged and the loop continues. Reply reads and capture retries are +paced on the monotonic clock, so a wall-clock step neither stalls nor rushes +them. Wall time stamps persistent records and governs their expiry and receipt +recovery. """ import calendar @@ -230,8 +236,9 @@ def epoch(text): class Pickup: - def __init__(self, home, root, courier_user, now=time.time): - self.home, self.now = home, now + def __init__(self, home, root, courier_user, now=time.time, clock=time.monotonic): + # clock keeps reply reads and capture retries independent of wall-clock steps. + self.home, self.now, self.clock = home, now, clock self.inbound = root / 'srv/courier/inbound' self.outbox = root / 'srv/courier/outbox' self.receipts = root / 'srv/courier/inbox' @@ -259,6 +266,7 @@ def __init__(self, home, root, courier_user, now=time.time): if self.s.get('version') != 1: raise Refused('unsupported pickup state') self.next_reply = 0.0 + self.retry_at = 0.0 # The pending capture's next attempt; a restart retries at once. self.quiet = {} # The turn this process last filed. Only this single-instance peer files # into the iMessage conversation, so it is the conversation's tail; a @@ -402,13 +410,14 @@ def take(self, seq, kind, key, record): 'created': record['published_at'], 'notice': None, 'asked': poll['asked'], 'binding': poll['binding'], 'question': poll['question']} self.s['polls'].pop(record['request_id']) - self.s['pending'].update(attempts=0, not_before=0, turn=None, request=None, previous=None) + self.retry_at = 0.0 + self.s['pending'].update(attempts=0, turn=None, request=None, previous=None) self.save() def file(self): """Try the pending capture once it is due; whether the queue moved past it.""" entry = self.s['pending'] - if entry['not_before'] > self.now(): + if self.retry_at > self.clock(): return False try: if entry['turn'] is None: @@ -430,7 +439,7 @@ def file(self): self.tail = None entry['attempts'] += 1 if entry['attempts'] <= len(RETRY_DELAYS): - entry['not_before'] = self.now() + RETRY_DELAYS[entry['attempts'] - 1] + self.retry_at = self.clock() + RETRY_DELAYS[entry['attempts'] - 1] log('filing %s failed (attempt %d): %s' % (entry['message'], entry['attempts'], exc)) self.save() return False @@ -589,7 +598,8 @@ def send(self): if entry['mark'] and entry['mark'][1] == 'question' else {}) result = self.receipt(entry['id']) if result is None and (entry['published'] is None - or (self.now() - entry['published'] >= RECEIPT_RETRY + # A publish time ahead of now means the clock stepped back. + or (not 0 <= self.now() - entry['published'] < RECEIPT_RETRY and not os.path.lexists(self.outbox / (entry['id'] + '.json')))): row = entry.get('wire') if row is None: @@ -647,8 +657,8 @@ def tick(self, force_replies=False): self.clear('pick') except (Refused, OSError, ValueError, KeyError, TypeError, subprocess.TimeoutExpired) as exc: self.once('pick', 'inbound spool unavailable: %s' % exc) - if self.s['pending'] is None and (force_replies or self.now() >= self.next_reply): - self.next_reply = self.now() + REPLY_EVERY + if self.s['pending'] is None and (force_replies or self.clock() >= self.next_reply): + self.next_reply = self.clock() + REPLY_EVERY try: self.replies() self.clear('replies') diff --git a/bin/fm_inbox_conversation.py b/bin/fm_inbox_conversation.py index 76fb0b70a0f..d3e12281c3a 100644 --- a/bin/fm_inbox_conversation.py +++ b/bin/fm_inbox_conversation.py @@ -32,13 +32,15 @@ therefore needs no reset of state/voice-conversation/policy.json. bind (owner): {conversation_id, authenticated_principal, destination?}. Returns - a random transport credential bound to this conversation and its authenticated - principal. destination defaults to "elevenlabs" and must be one the live - policy authorizes (a lab authorizes only the default); it is fixed at the - first bind, and publish refuses any reply for another destination, so a + {conversation_id, credential, destination?}: a random transport credential + bound to this conversation and its authenticated principal, plus the bound + destination when it is not the default. Transport calls carry only the + conversation_id and credential. destination defaults to "elevenlabs" and + must be one the live policy authorizes (a lab authorizes only the default); + it is fixed at the first bind, and publish refuses any reply for another destination, so a reply meant for a call never goes out as a text and a text never reaches a call. A conversation bound to a - destination other than "elevenlabs" reports it from accept and audit; one + destination other than "elevenlabs" reports it from bind, accept and audit; one bound to the default carries no such field, exactly as before it existed. Repeat bind by the session holding the lock is idempotent; a different principal or destination cannot adopt the conversation. @@ -362,7 +364,7 @@ def bind(self): self.c['destination'] = destination self.j['conversations'][self.cid] = self.c self.save() - return {'conversation_id': self.cid, 'credential': self.c['credential']} + return self.scoped({'conversation_id': self.cid, 'credential': self.c['credential']}) def wake(self, key): env = dict(os.environ, FM_HOME=str(self.home)) diff --git a/tests/fm-courier-pickup-cases.py b/tests/fm-courier-pickup-cases.py index 17f716c0ceb..3148b759aff 100644 --- a/tests/fm-courier-pickup-cases.py +++ b/tests/fm-courier-pickup-cases.py @@ -143,7 +143,8 @@ def courier_takes(receipted=True): binding = conversation('bind', {'conversation_id': 'text', 'authenticated_principal': 'captain', 'destination': 'imessage'}) (home / 'state/imessage').mkdir(mode=0o700) -(home / 'state/imessage/binding.json').write_text(json.dumps(dict(binding, destination='imessage'))) +assert binding['destination'] == 'imessage', binding +(home / 'state/imessage/binding.json').write_text(json.dumps(binding)) # exactly as bind returned it for directory in (inbound, outbox, receipts, courier / 'etc/courier'): directory.mkdir(parents=True) (courier / 'etc/courier/policy.toml').write_text( @@ -467,6 +468,9 @@ def courier_takes(receipted=True): inflight_receipt.unlink() once(clock_offset=2 * 86400 + 61) assert (outbox / (original['id'] + '.json')).read_bytes() == original_bytes +courier_takes(receipted=False) +once(clock_offset=2 * 86400 + 61 - 600) # The wall clock stepped back 10 minutes. +assert (outbox / (original['id'] + '.json')).read_bytes() == original_bytes assert [row['id'] for row in courier_takes()] == [original['id']] once() assert [row['text'] for row in courier_takes()] == ['Recovery complete.'] @@ -474,7 +478,7 @@ def courier_takes(receipted=True): assert ('receipt-recovery', 'done') in stages(), stages() assert all(row['delivery']['state'] == 'completed' for row in conversation('poll', binding)['replies'] if row['response_id'].startswith('recovery-')) -print('PASS: a pruned receipt is recovered by byte-identical same-ID republication, limited to once per 60 s') +print('PASS: a pruned receipt is recovered by byte-identical same-ID republication, limited to once per 60 s, and not stalled by a backward clock step') message('best-effort-poll', 'Ask with an optional native poll.') once() @@ -700,12 +704,8 @@ def courier_takes(receipted=True): # A message nobody can file - no live session - is told the failure sentence after its retries. (home / 'state/.lock').write_text('99999999\n') message('m5', 'anyone there?') -for _ in range(3): +for _ in range(3): # Each run is a new process, so its retry is due at once. once() - state = json.loads(state_path.read_text()) - if state['pending'] is not None: - state['pending']['not_before'] = 0 # skip the retry wait, not the retry - state_path.write_text(json.dumps(state)) assert json.loads(state_path.read_text())['pending'] is None assert ('m5', 'failed') not in stages(), stages() assert [row['text'] for row in courier_takes()] == ["I couldn't reach Firstmate."] @@ -723,8 +723,6 @@ def courier_takes(receipted=True): if attempt < 2: assert state['pending']['words'] == 'x' * 16001 assert state['pending']['attempts'] == attempt + 1 - state['pending']['not_before'] = 0 - state_path.write_text(json.dumps(state)) assert state['pending'] is None assert not any(r['turn_id'] == 'imsg-m-long' for r in requests()), requests() assert ('m-long', 'failed') not in stages(), stages() @@ -744,10 +742,6 @@ def courier_takes(receipted=True): 'digest': 'd', 'poll_message_id': 'p3'}) for _ in range(3): once() - state = json.loads(state_path.read_text()) - if state['pending'] is not None: - state['pending']['not_before'] = 0 - state_path.write_text(json.dumps(state)) assert json.loads(state_path.read_text())['pending'] is None assert [row['text'] for row in courier_takes()] == ["I couldn't reach Firstmate."] once() @@ -779,6 +773,33 @@ def courier_takes(receipted=True): assert second.returncode == 0 print('PASS: each record reaches the wake queue in under a second while the pickup runs') +# A backward wall-clock step while running neither stalls reading replies nor sending them. +stepped = ('import runpy,sys,time; wall=time.time; start=time.monotonic(); ' + 'time.time=lambda: wall()-(600 if time.monotonic()-start > 1 else 0); sys.argv=sys.argv[1:]; ' + 'runpy.run_path(sys.argv[0], run_name="__main__")') +while conversation('accept', {'conversation_id': 'text'})['dispatch']: + pass +running = subprocess.Popen([sys.executable, '-c', stepped, str(pickup), 'run'], env=env, stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, text=True) +try: + time.sleep(2.5) # past the 10-minute step back + conversation('publish', {'conversation_id': 'text', 'request_id': 'imsg-req-lat0', 'response_id': 'stepped', + 'sequence': 1, 'kind': 'answer', 'final': True, 'destination': 'imessage', + 'speech_text': 'Sent despite the clock.'}) + started = time.monotonic() + while not outgoing('.json'): + assert time.monotonic() - started < 10, 'a backward clock step stalled the reply' + assert running.poll() is None, 'pickup exited' + time.sleep(0.1) + assert [row['text'] for row in courier_takes()] == ['Sent despite the clock.'] + while ('lat0', 'done') not in stages(): + assert time.monotonic() - started < 10, 'the sent reply never moved its stage' + time.sleep(0.1) +finally: + running.terminate() + running.wait(timeout=10) +print('PASS: after a backward wall-clock step the running pickup still reads and sends replies on its cadence') + # Slow failure: a hung capture queues its notice, whose sent receipt precedes the # failed stage, inside the courier's 120 s PICKUP_DEADLINE from the record's rename. hang = open(home / 'state/voice-conversation/lock', 'a') diff --git a/tests/fm-inbox-conversation-cases.py b/tests/fm-inbox-conversation-cases.py index b95444d5190..c0f8c5fe2f5 100644 --- a/tests/fm-inbox-conversation-cases.py +++ b/tests/fm-inbox-conversation-cases.py @@ -445,6 +445,9 @@ def saved(cid='calls'): text_bind = {'conversation_id': 'text', 'authenticated_principal': 'captain', 'destination': 'imessage'} text = run('bind', text_bind) assert run('bind', text_bind) == text +assert set(text) == {'conversation_id', 'credential', 'destination'} and text['destination'] == 'imessage' +assert set(call) == {'conversation_id', 'credential'} # a default binding is unchanged +text = {k: text[k] for k in ('conversation_id', 'credential')} # the transport identity for rebind in (dict(text_bind, destination='elevenlabs'), {'conversation_id': 'text', 'authenticated_principal': 'captain'}, {'conversation_id': 'call', 'authenticated_principal': 'captain', 'destination': 'imessage'}, {'conversation_id': 'fax', 'authenticated_principal': 'captain', 'destination': 'fax'}):