(
events: readonly RuntimeEvent[],
options: { requireTerminalWrite?: boolean; requireDurableWrite?: boolean } = {},
)
| 920 | } |
| 921 | |
| 922 | async recordRuntimeEvents( |
| 923 | events: readonly RuntimeEvent[], |
| 924 | options: { requireTerminalWrite?: boolean; requireDurableWrite?: boolean } = {}, |
| 925 | ): Promise<void> { |
| 926 | if (events.length === 0) return; |
| 927 | for (const event of events) { |
| 928 | const terminal = isTerminalRuntimeEvent(event); |
| 929 | const eventForStore = terminal ? this.reserveTerminalEvent(event) : event; |
| 930 | if (!eventForStore) continue; |
| 931 | if (!this.input.runtimeEventStore || !this.runtimeEventStoreAvailable) { |
| 932 | if (this.input.runtimeEventStore?.durability === 'canonical') { |
| 933 | throw ( |
| 934 | this.runtimeEventStoreFailure ?? |
| 935 | new Error('canonical RuntimeEvent store is unavailable') |
| 936 | ); |
| 937 | } |
| 938 | if (terminal && options.requireTerminalWrite) { |
| 939 | throw new Error('terminal RuntimeEvent store is unavailable'); |
| 940 | } |
| 941 | if (options.requireDurableWrite && this.input.runtimeEventStore) { |
| 942 | // The store exists but earlier writes failed: a durability-required |
| 943 | // event (steering) must not silently skip the ledger. |
| 944 | throw new Error('RuntimeEvent store is unavailable for a durability-required event'); |
| 945 | } |
| 946 | continue; |
| 947 | } |
| 948 | const write = this.enqueueRuntimeEventStore( |
| 949 | 'append runtime event', |
| 950 | async () => { |
| 951 | await this.input.runtimeEventStore?.appendRuntimeEvent( |
| 952 | this.sessionId, |
| 953 | this.runId, |
| 954 | eventForStore, |
| 955 | { durable: terminal || options.requireDurableWrite === true }, |
| 956 | ); |
| 957 | }, |
| 958 | { |
| 959 | rethrow: |
| 960 | terminal || |
| 961 | options.requireTerminalWrite || |
| 962 | options.requireDurableWrite || |
| 963 | this.input.runtimeEventStore.durability === 'canonical', |
| 964 | }, |
| 965 | ); |
| 966 | if (terminal && this.terminalClaim) this.terminalClaim.write = write; |
| 967 | if (options.requireDurableWrite && !terminal) { |
| 968 | // An append error is AMBIGUOUS: the bytes may have landed before the |
| 969 | // failure (e.g. a close error after the write). For a |
| 970 | // durability-required event the caller settles a delivery lease on |
| 971 | // this outcome, so a false "not durable" would redeliver a message |
| 972 | // the ledger already owns. Read the ledger back to disambiguate: |
| 973 | // present ⇒ durable (continue on the ack path); absent or read-back |
| 974 | // also failing ⇒ fail closed (rethrow ⇒ nack). |
| 975 | try { |
| 976 | await write; |
| 977 | } catch (error) { |
| 978 | if (error instanceof DurableStoreWriteError) throw error; |
| 979 | if (!(await this.eventLandedInLedger(eventForStore.id))) throw error; |
no test coverage detected