fix chat event bus error recovery and subscriber error handling

- recovering from an error state now re-runs init() when latestRev was
  never seeded (previously polled forever with an undefined cursor,
  silently skipping all events), and resumes from the existing cursor
  when one exists (previously discarded it and skipped the offline gap)
- emit log batches outside poll()'s try/catch so a throwing subscriber
  is logged instead of being misreported as a network/poll failure

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Samuel Newman
2026-06-12 10:36:28 +03:00
parent ec695a42c5
commit e475a1ca36
+47 -17
View File
@@ -211,17 +211,9 @@ export class MessagesEventBus {
}
case MessagesEventBusStatus.Error: {
switch (action.event) {
case MessagesEventBusDispatchEvent.UpdatePoll: {
// basically reset
this.status = MessagesEventBusStatus.Initializing
this.latestRev = undefined
this.init()
break
}
case MessagesEventBusDispatchEvent.UpdatePoll:
case MessagesEventBusDispatchEvent.Resume: {
this.status = MessagesEventBusStatus.Ready
this.resetPoll()
this.emitter.emit('event', {type: 'connect'})
this.recoverFromError()
break
}
}
@@ -238,6 +230,31 @@ export class MessagesEventBus {
})
}
private recoverFromError() {
logger.debug(`recoverFromError`, {hasRev: !!this.latestRev})
if (this.latestRev === undefined) {
/*
* init() never succeeded, so we have no cursor to resume from. Re-run
* init() to seed latestRev. Its success path dispatches Ready, which from
* Initializing transitions us to Ready + resetPoll + emit connect.
*/
this.status = MessagesEventBusStatus.Initializing
this.init()
} else {
/*
* A poll failed mid-session but we still have a valid cursor. Resume
* polling from it directly. We must NOT route through init() here: its
* seeding logic takes the max of the existing rev and the server's
* current cursor, which would skip any events that arrived while we were
* offline.
*/
this.status = MessagesEventBusStatus.Ready
this.resetPoll()
this.emitter.emit('event', {type: 'connect'})
}
}
private async init() {
logger.debug(`init`, {})
@@ -337,6 +354,9 @@ export class MessagesEventBus {
// },
// )
let needsEmit = false
let batch: ChatBskyConvoGetLog.OutputSchema['logs'] = []
try {
const response = await networkRetry(2, () => {
return this.agent.chat.bsky.convo.getLog(
@@ -351,9 +371,6 @@ export class MessagesEventBus {
const {logs: events} = response.data
let needsEmit = false
let batch: ChatBskyConvoGetLog.OutputSchema['logs'] = []
for (const ev of events) {
/*
* If there's a rev, we should handle it. If there's not a rev, we don't
@@ -373,10 +390,6 @@ export class MessagesEventBus {
}
}
}
if (needsEmit) {
this.emitter.emit('event', {type: 'logs', logs: batch})
}
} catch (e: any) {
if (!isNetworkError(e) && !isErrorMaybeAppPasswordPermissions(e)) {
logger.error(`poll events failed`, {
@@ -397,5 +410,22 @@ export class MessagesEventBus {
} finally {
this.isPolling = false
}
/*
* Emit outside the try/catch above so a throwing subscriber is not
* misreported as a poll failure (which would show a network-error banner
* and drop the batch, since the revs have already been consumed). poll()
* runs from setInterval, so we must not let the exception escape - log it
* and move on without dispatching Error.
*/
if (needsEmit) {
try {
this.emitter.emit('event', {type: 'logs', logs: batch})
} catch (e) {
logger.error(`subscriber error handling chat events`, {
safeMessage: e instanceof Error ? e.message : String(e),
})
}
}
}
}