diff --git a/README.md b/README.md index a95a685..b207d88 100644 --- a/README.md +++ b/README.md @@ -1,24 +1,212 @@ -# Learning Observer Event Library +# lo_event — Learning Observer Event Library -This is a module used to stream events into the Learning Observer (and, in the future, potentially other Learning Record Stores). This is in development. The requirements are: +The 10,000-foot goal: **pipe diverse events from many sources into +learning record stores, including the original Learning Observer.** +Learning Observer is pluggable learning analytics; `lo_event` is the +client-side pipe that mates with it. -- We would like to be able to stream events with multiple loggers. - - In most cases, in practice, we use a websocket logger, with a persistent connection. - - For occasional events, we support AJAX logging. - - In addition, for ease-of-debugging, we can print events to the console - - We are beginning to support a workflow with `react` integration, which provides for very good observability -- We follow the general format used in Caliper, xAPI, and Open edX of one JSON object per event -- We use a free-form JSON format, but encourage following Caliper / xAPI guidelines where convenient -- We currently support JavaScript, but would like to support other languages in the future +It was built to feed Learning Observer from a handful of +sources. Writing Observer — a $2M IES-funded writing-process research +platform — was the flagship, alongside a few smaller research +prototypes. Since then it has grown to serve a different kind of +source as well: systems designed from the ground up to elicit and +surface student thinking through process data. lo-blocks is an +example. -Examples of places where we intentionally diverge from standards: +It does both: When a system is designed to provide good data, the +redux-style loop (below) gives us a *guarantee* of it: full +application state reconstructable from the stream. However, plenty of +systems aren't designed that way, and taking in their data is +messy. The data is partial: whatever they happen to emit. Pluggable +learning analytics means meeting sources where they are, and +correlating what you get. -- Good events are like onions -- they have layers. We don't assume we can e.g. trust timestamps or authentication from the system generating events, or that we will have all context up-front. Systems can add timestamps, authentication, and similar, much like e.g. SMTP messages being passed between systems. -- We do need to have a header for metadata + authentication -- We'd like to be at least sensitive to bandwidth. It's not worth resending data with each event that can be in a header or in update events. A lot of standards have large, cumbersome events (which are not human-friendly, and expensive to store and process) -- We're a lot more freeform in what we send and accept, since learning contexts can be pretty rich (and technology evolves) in ways which standards don't always keep up with. +## What we want from the library -Our goal is to simplify compatibility and to maintain compliance where reasonable, but to be more flexible than strict compliance with xAPI or Caliper. +- **Stream events through multiple loggers.** + - In practice, usually a **websocket logger** with a persistent connection. + - **AJAX logging** for occasional events. + - A **console logger**, because being able to see your events is + helpful for debugging. + - And increasingly a **`react`/`redux` integration**, which gives you very good + observability for free (your app state is your event log; see below). +- **One JSON object per event**, one line per event — the general shape used by + Caliper, xAPI, and Open edX. +- **Free-form JSON**, but follow Caliper / xAPI vocabulary where it's convenient. + Compatibility where reasonable; flexibility where learning outpaces the specs. +- **JavaScript today**, other languages later. + +The goal is to *simplify* compatibility and stay compliant where it's reasonable +— while being a good deal more flexible than strict xAPI or Caliper. + +## Events are like onions — they have layers + +We don't assume we can trust the timestamps or authentication of the system +generating an event, or that we'll have all the context up front. Systems add +timestamps, authentication, and context as the event is passed along — much like +an SMTP message picking up headers as it hops between servers. + +This isn't abstract. Some Tincan/xAPI libraries crash hard if an event +is missing a user, a timestamp, and so on. In practice, the client +generating the event often *can't* know the authenticated user; the +**server** ought to stamp it. And there isn't one true timestamp — +there's the browser clock, the JS server's receive time, the Python +server's receive time. We keep those as layers rather than pretending +a single authoritative time exists, and we let each system add what it +knows when it knows it. + +Consequences of taking layers seriously: + +- **There's a header for metadata + authentication.** Context that's the same + across many events (source, version, the authenticated identity) rides the + header, stamped once, not re-sent per event. +- **We're sensitive to bandwidth.** It's not worth resending, on every + event, what can live in a header or in an occasional update event. A + lot of standards ship large, cumbersome events — not human-friendly, + and expensive to store and process. So context is sent **once** (via + `lock_fields`) and **omitted from subsequent events** until it + changes; the server denormalizes it back per event. Downstream, + locking in a page URL or sending it with each event are equivalent + (we denormalize when processing). +- **We're freeform in what we send and accept.** Learning contexts are rich, and + the technology evolves in ways the standards don't always keep up with. + +## Two ways people use it + +**A. Fire-and-forget telemetry (the original).** Configure loggers, +call `logEvent(...)`, events stream out. This is how a writing +extension or any external system feeds Learning Observer. The system +tries very hard never to lose an event (see *The delivery standard*). + +**B. redux application state (the expanded role).** Events flow *through* a redux +store, so application state and event logging are one data flow — which is why +you get observability for free. The client applies its own events optimistically; +the server folds the *same* event stream through the *same* reducers into +authoritative state, and can push events back down. This is how lo-blocks builds +event-sourced, collaborative activities. + +The two are not fully exclusive; in some cases, it is convenient to +add additional events on top of redux. + +In the observable mode, on the client, **redux is the event bus** — +folding every change through reducers is what makes full application +state reconstructable from the event stream. Current state can be +thought of as a cache of the event stream. + +We are gradually moving towards a model where **the server acts like +redux** as well: fast local redux for UX, a slower authoritative +server-side redux-equivalent for shared state. `lo_event` is the +substrate both sit on (format, durable queue, transport, acks, client +redux front-end). It is a *sync-aware transport with a redux +front-end*, **not** a distributed state engine — server-side folding +is the consumer's business. + +## The delivery standard + +The event stream is the ground source of truth. Losing an event means losing +student work. The hard promise starts when the outbox write commits: from then +on, the event is always either held by the client or captured by the server. +Three layers, and **only the bottom is hard**: + +1. **Capture (hard).** `logEvent → in-memory front desk → durable outbox`. No + network gate participates in admission. An awaitable admission API that can + surface IndexedDB failure is future work. +2. **Transmission (soft).** A lease decides *when* to send and an explicit + identity ack decides *when to delete* — only after server capture. + Un-acked events sit durably and resend on reconnect. Nothing is dropped, only + deferred. +3. **UX (soft).** Hooks (`useConnected`, `useSaved`, …) let the app + disable the interface. Presentation only — never a reason an event + isn't captured -- but the goal is to be able to stop sending events + if we lost connection (tell the user we're not connected). Of + course, for offline operation, we would not use this; events would + simply restream on reconnect. There is a time for both. + +"Soft shut down the flow, keep hard delivery" = throttle layer 2, disable layer +3, **never touch layer 1**. A consumer that disables the UI on failure must do it +**read-only-after-capture** (log the event, *then* lock the input), never drop +pending input. + +## Durability & the ack protocol + +Backed by **IndexedDB** (the browser default), the queue survives reloads and +outages — events persist across a page reload or process restart. The +**in-memory fallback** (Node, or where IndexedDB is unavailable) survives +in-process outages and reconnects, but not a reload/restart. Reliability rests +on a **lease discipline**, not delete-on-read: + +- Every stored event gets a monotonic storage id (`seq`) used only inside its + browser queue. It is never sent on the wire. +- **Lease, don't take:** `leaseNext()` hands an item to the sender without + deleting it; `confirm(seqs)` deletes exactly the named storage records; + `rewind()` re-hands everything unacknowledged on reconnect. +- Each frame carries an opaque `metadata.eventId`. The server replies with + `{ status: 'ack', id: eventId }` after capturing it. The sender maps that + identity back to the exact storage record it sent. +- **At-least-once, not exactly-once.** Reducers must tolerate duplicates and + reordering; concurrently editable fields use CRDTs. Resending is always safer + than deleting work without proof. + +Why it exists: the old queue deleted an event in the same transaction it read it +for sending, and `socket.send()` is fire-and-forget — so an event closed-tab in +that window vanished silently (we measured real tail-of-session losses). The +lease discipline closes that window. + +There is no capability negotiation. `autoack: false` (the default) is durable: +only a server ack confirms a named record. `autoack: true` is an explicit +send-and-forget profile that confirms after a send on a verified-OPEN socket. +A durable client pointed at an ack-less server retains a visible backlog rather +than silently dropping work. + +## Client state-sync (the redux workflow) + +The server can also send events *down* for the client's reducers, addressed by +`state://` keys and delivered by subscription **bound to the connection**. One +bus, three multiplexed traffic classes ("planes"): + +| Plane | Direction | Carries | Reliability | +|---|---|---|---| +| 1 — client events | client→server | durable user actions | server-acked, resend on reconnect | +| 2 — control | client→server | `subscribe` / `unsubscribe` (batched) | idempotent, re-sent each connect, no ack | +| 3 — server events | server→client | events for the client's reducers | client-acked (ordering only); recovery = snapshot on (re)subscribe | + +**No echo, either direction.** The client applies its own events optimistically; +the server folds authoritatively and forwards only to *other* subscribers. +Recovery is the **snapshot returned on (re)subscribe**, not an echo — so acks +carry no state; they exist only for durability tracking and replay ordering. + +Inbound server events go through the same reducer registry as local ones: a +set-the-value reducer for the simple case, a registered merge reducer (e.g. CRDT) +for the hard case. The server folds with the same reducer code — a property of +the consumer (lo-blocks), not something `lo_event` encodes. The full wire +contract lives in the consuming project's protocol doc; `lo_event` implements the +client half. + +## React hooks + +Import from `lo_event/hooks` (this entry needs React). They read a plain +module-level status store via `useSyncExternalStore` — **not** Redux state — so +consumers stay reactive without an imperative `consumeCustomEvent` listener. + +| Hook | Returns | Meaning | +|---|---|---| +| `useConnected()` | `true \| false \| null` | connected / offline / no websocket configured | +| `useSaved()` | `'saved' \| 'modified' \| 'error'` | persistence status | +| `useLoaded()` | `boolean` | initial state resolved (gate the UI on this) | + +## Failure handling + +When something fails in a way worth noticing, log to **all three** of: + +1. **The console** (`debug.error`) — always. +2. **localStorage** (`util.recordFailure`) — a bounded, ring-buffered NDJSON log + (`lo_event_failures`). Sized by calculation, not vibes: capped at a small + fixed slice (~128 K chars ≈ ~5% of the ~5 MB browsers guarantee) and + ring-buffered, so it can never grow toward the quota. Deliberately **not** + wired to every `debug.error` — call it only for failures worth persisting, or + you get exponentially growing logs. +3. **A consumer surface** — connection, loaded, saved, and durable-queue status + let the application report trouble without throwing from the logging path. ## Installation @@ -26,68 +214,93 @@ Our goal is to simplify compatibility and to maintain compliance where reasonabl npm install ``` -To use in a separate node project: +For local development against another project, link the checkout (or see +`pack-install` in `package.json` for a tarball-based alternative that sidesteps +`npm link` quirks). -```bash -npm install -npm link -``` +## Usage -Then from the other project: +The basic loop is four calls — configure loggers, lock in context, start +streaming, log events: -```bash -npm link lo-event -``` +```js +import * as lo_event from 'lo_event'; +import { consoleLogger } from 'lo_event/console'; +import { websocketLogger } from 'lo_event/websocket'; -*Note:* you may need to rerun `npm link lo-event` after you run `npm install` at the target location. +// 1. Configure loggers. Each event fans out to all of them. A logger is just a +// function that receives a JSON-encoded event string, so it's easy to add +// your own (console for dev, websocket for the persistent connection, AJAX +// for occasional events, …). +lo_event.init('my-app', '1.0.0', [ + consoleLogger(), + websocketLogger('wss://example.org/wsapi/in/', { + autoack: false, + namespace: 'my-app', + }), +]); -If this runs into issues, a more robust way is to run `npm pack` to create a tarball npm package, and then to `npm install` that package. This has the downside of requiring a reinstall on every change, which is somewhat cumbersome. +// 2. Optional: lock in context that rides the header — sent once, denormalized +// back onto each event server-side, not re-sent per event. +lo_event.lockFields([{ course: 'greenheart', activity: 'field-guide' }]); -## Usage +// 3. Start streaming. Anything logged (or locked) before go() is queued and +// sent first, in order. +lo_event.go(); -```js -import * as lo_event from 'lo-event'; -import { consoleLogger } from 'lo-event/console'; -import { websocketLogger } from 'lo-event/websocket'; -import { reduxLogger } from 'lo-event/redux'; -import * as debug from 'lo-event/debug'; -import { subscribeToEvents } from 'lo-event/browser-events'; -import * as util from 'lo-event/util'; +// 4. Log events — one flat JSON object each. Follow xAPI/Caliper vocabulary +// where convenient; be freeform where it isn't. +lo_event.logEvent('SUBMIT', { problem: 'q1', correct: true }); ``` +The order matters: `init` → any pre-auth `lockFields` → `go` → `logEvent`. +Events logged before `go()` don't get dropped — they queue durably and stream +once `go()` runs. And every `logEvent` lands in the **durable queue first**; the +websocket logger streams it and (in ack mode) holds it until the server +confirms, so a reload or outage doesn't lose it. That's what "tries very hard +never to lose an event" means in practice — see *The delivery standard* and +*Durability & the ack protocol*. + +This mode is plain JavaScript — no React or redux required. For the +state-sourced workflow, add `reduxLogger` and the `lo_event/hooks` (see +*Client state-sync* and *React hooks*). + ## Exports | Specifier | Module | |---|---| -| `lo-event` | Main entry point (`loEvent.js`) | -| `lo-event/redux` | Redux logger integration | -| `lo-event/debug` | Debug logging utilities | -| `lo-event/console` | Console logger | -| `lo-event/websocket` | WebSocket logger | -| `lo-event/browser-events` | Browser event capture | -| `lo-event/queue` | Event queue | -| `lo-event/storage` | Browser storage abstraction | -| `lo-event/disabler` | Opt-in/opt-out handling | -| `lo-event/util` | Utility functions | -| `lo-event/null` | Null logger (no-op) | +| `lo_event` | Main entry point (`loEvent.js`) | +| `lo_event/redux` | Redux logger + reducer registry | +| `lo_event/hooks` | React status hooks (needs React) | +| `lo_event/websocket` | WebSocket logger (durable queue, ack protocol) | +| `lo_event/console` | Console logger | +| `lo_event/browser-events` | Browser event capture | +| `lo_event/queue` | Event queue (lease / confirm / rewind) | +| `lo_event/storage` | Browser storage abstraction | +| `lo_event/disabler` | Opt-in / opt-out handling | +| `lo_event/debug` | Debug logging utilities | +| `lo_event/util` | Utility functions (incl. `recordFailure`) | +| `lo_event/types` | Shared TypeScript types | +| `lo_event/null` | Null logger (no-op) | ## Examples -The `examples/` directory has interactive browser demos: +The `examples/` directory has interactive browser demos (`npm run browser`): -- **Browser Events** (`browser_events.html`) — Captures keystrokes, mouse, clipboard, and other DOM events using `subscribeToEvents`. Shows how metadata collectors work. -- **Redux Loop** (`redux_loop.html`) — Demonstrates the Redux logger, where events flow through a Redux store so application state and event logging share one data flow. - -To run them: - -```bash -npm run browser -``` - -This starts a Parcel dev server and opens the example index page. +- **Browser Events** (`browser_events.html`) — keystrokes, mouse, clipboard via + `subscribeToEvents`; shows how metadata collectors work. +- **Redux Loop** (`redux_loop.html`) — the redux logger, where events flow + through a store so application state and logging share one data flow. ## Testing ```bash npm test ``` + +Testing philosophy: good tests > no tests > bad tests. The system is inherently +testable — wire events through reducers, render example files, keep assertions +declarative. Avoid committed mocks/harnesses/polyfills (each is one more thing to +keep aligned with the code); interim scaffolding stays uncommitted, and tests of +a now-stable algorithm are weighed against their maintenance cost before they're +kept. diff --git a/package-lock.json b/package-lock.json index 77c4a94..5e96155 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "lo_event", - "version": "0.0.5", + "version": "0.0.9-ack.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "lo_event", - "version": "0.0.5", + "version": "0.0.9-ack.0", "license": "SEE LICENSE IN LICENSE.TXT", "dependencies": { "lodash": "^4.17.21", @@ -24,6 +24,7 @@ "@types/ws": "^8.18.1", "@typescript-eslint/parser": "^8.55.0", "eslint": "^9.39.2", + "fake-indexeddb": "^6.2.5", "globals": "^17.3.0", "parcel": "^2.12.0", "react": "^19.2.4", @@ -31,6 +32,14 @@ "tsup": "^8.5.1", "typescript": "^5.9.3", "vitest": "^3.0.0" + }, + "peerDependencies": { + "react": ">=18.0.0" + }, + "peerDependenciesMeta": { + "react": { + "optional": true + } } }, "node_modules/@babel/runtime": { @@ -4309,6 +4318,16 @@ "node": ">=12.0.0" } }, + "node_modules/fake-indexeddb": { + "version": "6.2.5", + "resolved": "https://registry.npmjs.org/fake-indexeddb/-/fake-indexeddb-6.2.5.tgz", + "integrity": "sha512-CGnyrvbhPlWYMngksqrSSUT1BAVP49dZocrHuK0SvtR0D5TMs5wP0o3j7jexDJW01KSadjBp1M/71o/KR3nD1w==", + "dev": true, + "license": "Apache-2.0", + "engines": { + "node": ">=18" + } + }, "node_modules/fast-deep-equal": { "version": "3.1.3", "resolved": "https://registry.npmjs.org/fast-deep-equal/-/fast-deep-equal-3.1.3.tgz", diff --git a/package.json b/package.json index a42490a..d1046c8 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "lo_event", - "version": "0.0.5", + "version": "0.0.9-ack.0", "description": "Event logging library for the Learning Observer", "main": "dist/loEvent.js", "types": "dist/loEvent.d.ts", @@ -61,10 +61,23 @@ "./types": { "types": "./dist/types.d.ts", "import": "./dist/types.js" + }, + "./hooks": { + "types": "./dist/hooks.d.ts", + "import": "./dist/hooks.js" + } + }, + "peerDependencies": { + "react": ">=18.0.0" + }, + "peerDependenciesMeta": { + "react": { + "optional": true } }, "scripts": { "test": "vitest run", + "typecheck": "tsc --noEmit", "test:watch": "vitest", "build": "tsup", "prepublishOnly": "npm run build", @@ -100,6 +113,7 @@ "@types/ws": "^8.18.1", "@typescript-eslint/parser": "^8.55.0", "eslint": "^9.39.2", + "fake-indexeddb": "^6.2.5", "globals": "^17.3.0", "parcel": "^2.12.0", "react": "^19.2.4", diff --git a/src/disabler.ts b/src/disabler.ts index 716768f..b4ad6a8 100644 --- a/src/disabler.ts +++ b/src/disabler.ts @@ -100,6 +100,17 @@ export function streamEvents () { return action === EVENT_ACTION.TRANSMIT; } +/** Explicit predicates for the two permanent cases. Callers must never infer a + * privacy deletion from retry()'s boolean: permanent MAINTAIN means retain the + * outbox forever, while permanent DROP is the one sanctioned destructive case. */ +export function isPermanent (): boolean { + return expiration === TIME_LIMIT.PERMANENT; +} + +export function isPermanentOptOut (): boolean { + return isPermanent() && action === EVENT_ACTION.DROP; +} + /** * Determines if a client should retry based on the `expiration` status. * This function: @@ -108,14 +119,15 @@ export function streamEvents () { * initializations), and then returns `true` to allow a retry. */ export async function retry () { - if (expiration === TIME_LIMIT.PERMANENT) { - return false; - } - const now = Date.now(); - if (now < expiration!) { - debug.info(`waiting for expiration to happen ${new Date(expiration!).toString()}`); - await util.delay(expiration! - now); - debug.info('we are done waiting'); + while (true) { + if (expiration === TIME_LIMIT.PERMANENT) return false; + const deadline = expiration; + const now = Date.now(); + if (deadline === null || now >= deadline) break; + debug.info(`waiting for expiration to happen ${new Date(deadline).toString()}`); + await util.delay(deadline - now); + // A later block frame may have extended or made the block permanent while + // we slept. Re-read state instead of clearing that newer instruction. } action = DEFAULTS.action; expiration = DEFAULTS.expiration; diff --git a/src/hooks.ts b/src/hooks.ts new file mode 100644 index 0000000..509a1e5 --- /dev/null +++ b/src/hooks.ts @@ -0,0 +1,49 @@ +/** + * React hooks for lo_event persistence status. + * + * These use useSyncExternalStore against a plain module-level store + * in reduxLogger (NOT Redux state — see reduxLogger.ts for rationale). + * + * Import from 'lo_event/hooks' — this entry point depends on React. + */ +import { useSyncExternalStore } from 'react'; +import { + subscribeStatus, + getSaveStatus, + getConnected, + getLoaded, +} from './reduxLogger.js'; + +export type { SaveStatus } from './reduxLogger.js'; + +/** + * Whether the current state has been persisted. + * + * 'saved' — all changes have been sent to the server / localStorage + * 'modified' — changes exist, debounce timer running + */ +export function useSaved() { + return useSyncExternalStore(subscribeStatus, getSaveStatus, () => 'saved' as const); +} + +/** + * WebSocket connection status. + * + * true — connected + * false — disconnected (show offline indicator) + * null — no WebSocket configured (don't show indicator) + */ +export function useConnected() { + return useSyncExternalStore(subscribeStatus, getConnected, () => null); +} + +/** + * Whether initialization is complete (a fetch_blob load cycle has resolved). + * + * Use this to gate the UI — show a loading screen until true. Note: a store + * with no fetch_blob server never resolves loaded (local-only mode is not yet + * supported — see reduxLogger). + */ +export function useLoaded() { + return useSyncExternalStore(subscribeStatus, getLoaded, () => false); +} diff --git a/src/indexeddbQueue.ts b/src/indexeddbQueue.ts index 817b6a4..7a4fee5 100644 --- a/src/indexeddbQueue.ts +++ b/src/indexeddbQueue.ts @@ -1,252 +1,219 @@ -/** - * This files functions as a Queue using an indexeddb backend. - * - * If we are operating in a browser environment, we will use - * the built-in indexeddb. In node environments, we will use - * packages that mirror the functionality of indexeddb. - * - * Each item can be added to the end of the queue with `enqueue(item)`. - * Items can be retrieved from the queue with `item = await dequeue()`. - * - * TODO - * This code works in the browser, but breaks in a node environment. - * autoIncrement is NOT supported when working in the node - * environment. We will likely need to make some form of wrapper - * to achieve this behavior for node. - * See https://github.com/metagriffin/indexeddb-js/blob/master/src/indexeddb-js.js#L418C1-L418C53 - * NOTE: When we had our own counter for the id, we did notice that the node - * environment (indexeddb-js or sqlite3) handled keys differently, thus - * returning items out of order. - * - * TODO: This needs a very good code review. We weren't able to do - * this before merge. - */ import * as debug from './debugLog.js'; -import * as util from './util.js'; +import type { LeasedItem } from './types.js'; -const ENQUEUE = 'enqueue'; -const DEQUEUE = 'dequeue'; +interface StoredRecord { + id?: number; + payload: unknown; +} -interface DBOperation { - operation: string; - payload?: { payload: unknown }; - resolve?: (value: unknown) => void; - reject?: (reason?: unknown) => void; +const CROSS_CONTEXT_POLL_MS = 300; + +function transactionDone (transaction: IDBTransaction): Promise { + return new Promise((resolve, reject) => { + transaction.oncomplete = () => resolve(); + transaction.onerror = () => reject(transaction.error); + transaction.onabort = () => reject(transaction.error); + }); } +/** + * Durable, shared outbox. Records leave only through explicit-id confirm(). + * The lease cursor is per instance and deliberately ephemeral. + */ export class Queue { - private db: IDBDatabase | null; - private dbOperationQueue: DBOperation[]; - private nextDBOperationPromise: ((value: DBOperation) => void) | null; - private nextItemPromise: ((value: unknown) => void) | null; - private queueName: string; - private dbOperationDispatch: Record Promise>; - nextDBOperation: () => AsyncGenerator; - - constructor (queueName: string) { - this.db = null; - this.dbOperationQueue = []; - this.nextDBOperationPromise = null; - this.nextItemPromise = null; - this.queueName = queueName; - - this.initialize = this.initialize.bind(this); - this.addItemToDB = this.addItemToDB.bind(this); - this.nextItemFromDB = this.nextItemFromDB.bind(this); - this.nextDBOperation = util.once(this._nextDBOperation.bind(this)); - this.startProcessing = this.startProcessing.bind(this); - this.addItemToDBOperationQueue = this.addItemToDBOperationQueue.bind(this); - this.enqueue = this.enqueue.bind(this); - this.dequeue = this.dequeue.bind(this); - - this.dbOperationDispatch = { - [ENQUEUE]: this.addItemToDB, - [DEQUEUE]: this.nextItemFromDB - }; - this.initialize(); - } - - /** - * Determine which environment we are in to set - * the appropriate indexeddb information. - */ - async initialize () { - let request; + private readonly ready: Promise; + private writes: Promise = Promise.resolve(); + private leasedThrough = 0; + /** Changes whenever rewind/clear invalidates an asynchronous cursor result. */ + private leaseEpoch = 0; + private parkedLease: ((value: LeasedItem | PromiseLike) => void) | null = null; + private parkedLeaseTimer: ReturnType | null = null; + + constructor (private readonly queueName: string) { if (typeof indexedDB === 'undefined') { - // Node.js persistent queue is not yet supported. - // The sqlite3/indexeddb-js fallback was broken (autoIncrement - // unsupported, keys returned out of order) and the imports - // break browser bundlers. Use QueueType.IN_MEMORY for now. - // - // To restore Node support, install sqlite3 and indexeddb-js - // and uncomment: - // const sqlite3 = await import('sqlite3'); - // const indexeddbjs = await import('indexeddb-js'); - // const engine = new sqlite3.default.Database('queue.sqlite'); - // const scope = indexeddbjs.makeScope('sqlite3', engine); - // request = scope.indexedDB.open(this.queueName); - throw new Error( - 'IndexedDB is not available in this environment. ' + - 'Use QueueType.IN_MEMORY for Node.js.' - ); - } else { - debug.info('idbQueue: Using browser consoleDB'); - request = indexedDB.open(this.queueName, 1); + throw new Error('IndexedDB is not available. Use QueueType.IN_MEMORY outside a browser.'); } + this.ready = this.open(); + } - request.onerror = () => { - debug.error('QUEUE ERROR: could not open database', request.error); - }; + private open (): Promise { + return new Promise((resolve, reject) => { + const request = indexedDB.open(this.queueName, 1); + request.onupgradeneeded = () => { + request.result.createObjectStore(this.queueName, { keyPath: 'id', autoIncrement: true }); + }; + request.onsuccess = () => resolve(request.result); + request.onerror = () => reject(request.error); + request.onblocked = () => reject(new Error(`IndexedDB queue ${this.queueName} is blocked`)); + }); + } - request.onupgradeneeded = async () => { - this.db = request.result; - const objectStore = this.db.createObjectStore(this.queueName, { keyPath: 'id', autoIncrement: true }); - objectStore.createIndex('id', 'id'); - }; + /** Serialize local writes so a following local read observes them. A failed + * write is loud and does not poison later operations. The public logger API + * remains non-awaitable; surfacing admission failure is tracked in spec §11. */ + private scheduleWrite (write: () => Promise): void { + const operation = this.writes.then(write); + this.writes = operation.catch(error => { + debug.error(`IndexedDB queue ${this.queueName} write failed`, error); + }); + } - request.onsuccess = () => { - this.db = request.result; - this.startProcessing(); - }; + enqueue (item: unknown): void { + this.scheduleWrite(async () => { + const db = await this.ready; + const transaction = db.transaction(this.queueName, 'readwrite'); + transaction.objectStore(this.queueName).add({ payload: item } satisfies StoredRecord); + await transactionDone(transaction); + this.wakeParkedLease(); + }); } - /** - * Perform transaction to add item into indexeddb - * If we are waiting for an item to available to dequeue, - * we resolve the item immediately and don't add it to - * the indexeddb. - */ - async addItemToDB (op: DBOperation) { - const payload = op.payload!; - if (this.nextItemPromise) { - this.nextItemPromise(payload.payload); - this.nextItemPromise = null; - return; - } - debug.info(`idbQueue: adding item to database, ${payload}`); - const transaction = this.db!.transaction([this.queueName], 'readwrite'); - const objectStore = transaction.objectStore(this.queueName); + /** Scan using a captured cursor. The caller checks leaseEpoch before applying + * the result, so a rewind during this transaction cannot be overwritten. */ + private async scan (after: number): Promise { + await this.writes; + const db = await this.ready; + const transaction = db.transaction(this.queueName, 'readonly'); + const request = transaction.objectStore(this.queueName) + .openCursor(IDBKeyRange.lowerBound(after, true)); + const leased = await new Promise((resolve, reject) => { + request.onsuccess = () => { + const cursor = request.result; + resolve(cursor + ? { seq: cursor.key as number, item: (cursor.value as StoredRecord).payload } + : null); + }; + request.onerror = () => reject(request.error); + }); + await transactionDone(transaction); + return leased; + } - const request = objectStore.add(payload); + async leaseNext (): Promise { + while (true) { + const epoch = this.leaseEpoch; + const leased = await this.scan(this.leasedThrough); + if (epoch !== this.leaseEpoch) continue; + if (leased) { + this.leasedThrough = leased.seq; + return leased; + } - request.onsuccess = () => { - // successful request added - }; + return await new Promise(resolve => { + this.parkedLease = resolve; + this.scheduleParkedPoll(); + }); + } + } - request.onerror = () => { - if (request.error?.name === 'ConstraintError') { - debug.error('IDBQUEUE ERROR: Item already exists', request.error); - } else { - debug.error('IDBQUEUE ERROR: Error adding item to the queue:', request.error); + confirm (seqs: number[]): void { + if (!seqs.length) return; + this.scheduleWrite(async () => { + const db = await this.ready; + const transaction = db.transaction(this.queueName, 'readwrite'); + const store = transaction.objectStore(this.queueName); + for (const seq of new Set(seqs)) { + const request = store.delete(seq); + request.onerror = event => { + // Prevent one request error from aborting and rolling back its sibling + // deletes. Promise settlement belongs to the transaction (L15). + event.preventDefault(); + debug.error(`Unable to confirm IndexedDB record ${seq}`, request.error); + }; } - }; + await transactionDone(transaction); + }); } - /** - * Perform transaction to fetch next item in indexeddb - */ - async nextItemFromDB (op: DBOperation) { - const { resolve, reject } = op; - debug.info('idbQueue: Fetching next item from database'); - const transaction = this.db!.transaction([this.queueName], 'readwrite'); - const objectStore = transaction.objectStore(this.queueName); - const request = objectStore.openCursor(); - - request.onsuccess = () => { - const cursor = request.result; - if (cursor) { - const item = cursor.value; - const deleteRequest = objectStore.delete(cursor.key); - - deleteRequest.onsuccess = () => { - resolve!(item.payload); - }; + rewind (): void { + this.leaseEpoch++; + this.leasedThrough = 0; + this.wakeParkedLease(); + } - deleteRequest.onerror = () => { - debug.error('IDBQUEUE ERROR: Error removing item from the queue:', deleteRequest.error); - reject!(deleteRequest.error); - }; - } else { - // No more items in the IndexedDB. - resolve!(new Promise((resolve) => { - this.nextItemPromise = resolve; - })); - } - }; + async unconfirmedCount (): Promise { + await this.writes; + const db = await this.ready; + const transaction = db.transaction(this.queueName, 'readonly'); + const request = transaction.objectStore(this.queueName).count(); + return await new Promise((resolve, reject) => { + request.onsuccess = () => resolve(request.result); + request.onerror = () => reject(request.error); + }); + } - request.onerror = () => { - debug.error('IDBQUEUE ERROR: Error reading queue cursor:', request.error); - reject!(request.error); - }; + async maxSeq (): Promise { + await this.writes; + const db = await this.ready; + const transaction = db.transaction(this.queueName, 'readonly'); + const request = transaction.objectStore(this.queueName).openCursor(null, 'prev'); + return await new Promise((resolve, reject) => { + request.onsuccess = () => resolve(request.result ? request.result.key as number : null); + request.onerror = () => reject(request.error); + }); } - /** - * The processing loop continually waits for the next - * dbOperation to come using the following generator. - */ - private async * _nextDBOperation (): AsyncGenerator { + async unleasedAtOrBelow (seq: number): Promise { + // A failed send can rewind while this asynchronous count is in flight. + // Re-run against the new cursor rather than letting a pre-rewind zero clear + // the snapshot barrier with unsent backlog still present. while (true) { - let operation: DBOperation; - if (this.dbOperationQueue.length > 0) { - operation = this.dbOperationQueue.shift()!; - } else { - operation = await new Promise(resolve => { - this.nextDBOperationPromise = resolve; - }); - } - debug.info(`idbQueue: Yielding next operation, ${operation}`); - yield operation; + const epoch = this.leaseEpoch; + const leasedThrough = this.leasedThrough; + await this.writes; + if (epoch !== this.leaseEpoch) continue; + if (leasedThrough >= seq) return 0; + + const db = await this.ready; + const transaction = db.transaction(this.queueName, 'readonly'); + const range = IDBKeyRange.bound(leasedThrough, seq, true, false); + const request = transaction.objectStore(this.queueName).count(range); + const count = await new Promise((resolve, reject) => { + request.onsuccess = () => resolve(request.result); + request.onerror = () => reject(request.error); + }); + await transactionDone(transaction); + if (epoch === this.leaseEpoch) return count; } } - /** - * This method processes incoming dbOperations - */ - async startProcessing () { - const dbOperationStream = this.nextDBOperation(); - - for await (const operation of dbOperationStream) { - debug.info(`idbQueue: processing operation ${operation}`); - try { - await this.dbOperationDispatch[operation.operation](operation); - } catch (error) { - debug.error('Unable to perform operation on DB', error); - } - } + async inspect (limit = 20): Promise { + await this.writes; + const db = await this.ready; + const transaction = db.transaction(this.queueName, 'readonly'); + const request = transaction.objectStore(this.queueName).getAll(undefined, limit); + return await new Promise((resolve, reject) => { + request.onsuccess = () => resolve(request.result); + request.onerror = () => reject(request.error); + }); } - // helper function for enqueue/dequeue - addItemToDBOperationQueue (payload: DBOperation) { - if (this.nextDBOperationPromise) { - this.nextDBOperationPromise(payload); - this.nextDBOperationPromise = null; - } else { - this.dbOperationQueue.push(payload); - } + clear (): void { + this.scheduleWrite(async () => { + const db = await this.ready; + const transaction = db.transaction(this.queueName, 'readwrite'); + transaction.objectStore(this.queueName).clear(); + await transactionDone(transaction); + this.leaseEpoch++; + this.leasedThrough = 0; + }); } - /** - * This functions will append an enqueue message to the - * current operation stream. - */ - enqueue (item: unknown) { - debug.info(`idbQueue: Enqueuing item ${item}`); - const payload = { - operation: ENQUEUE, - payload: { payload: item } - }; - this.addItemToDBOperationQueue(payload); + /** IndexedDB has no portable cross-context change event. Every wake re-runs + * the normal scan, and a slow poll discovers records committed by other tabs. + * BroadcastChannel would only be an optimization; correctness stays here. */ + private scheduleParkedPoll (): void { + if (!this.parkedLease || this.parkedLeaseTimer !== null) return; + this.parkedLeaseTimer = setTimeout(() => this.wakeParkedLease(), CROSS_CONTEXT_POLL_MS); + (this.parkedLeaseTimer as ReturnType & { unref?: () => void }).unref?.(); } - /** - * This function appends a dequeue message to the operation - * stream and returns the result. - */ - dequeue () { - debug.info('idbQueue: dequeueing item'); - return new Promise((resolve, reject) => { - const payload = { operation: DEQUEUE, resolve, reject }; - this.addItemToDBOperationQueue(payload); - }); + private wakeParkedLease (): void { + if (!this.parkedLease) return; + if (this.parkedLeaseTimer !== null) clearTimeout(this.parkedLeaseTimer); + this.parkedLeaseTimer = null; + const resolve = this.parkedLease; + this.parkedLease = null; + resolve(this.leaseNext()); } } diff --git a/src/loEvent.ts b/src/loEvent.ts index eec9650..fdc8e61 100644 --- a/src/loEvent.ts +++ b/src/loEvent.ts @@ -3,6 +3,7 @@ */ import { timestampEvent, mergeMetadata } from './util.js'; +import type { QueueDebug } from './types.js'; import { getBrowserInfo } from './metadata/browserinfo.js'; import * as Queue from './queue.js'; import * as disabler from './disabler.js'; @@ -114,6 +115,97 @@ async function lockFieldsAsync (data: Record[]) { await Promise.all(authpromises); } +/** + * Total enqueued-but-unacked events across all ack-aware loggers (currently + * websocketLogger). Zero means every event has been durably acknowledged by + * the server — the precise "is anything unsaved?" signal for a beforeunload + * warning, replacing the blob-based heuristic. Loggers without ack support + * (which confirm on send) contribute zero. + */ +export async function unackedCount (): Promise { + const counts = await Promise.all( + loggersEnabled + .filter(logger => typeof logger.unackedCount === 'function') + .map(logger => Promise.resolve(logger.unackedCount!())) + ); + return counts.reduce((sum, n) => sum + n, 0); +} + +/** + * Console debugging for the durable queue. + * + * Attached to `globalThis.loDebug` in browsers, because the useful moment for + * this is a console prompt in a stuck tab, where there is no module to import: + * + * loDebug.queue() what is waiting, and WHY it is waiting + * loDebug.clearQueue() drop everything, unsent included + * + * `queue()` summarizes rather than dumping records. A queue that only grows + * looks identical whether the client is offline, the server is not acking, or + * a frame was enqueued that can never BE acked — and the last one is invisible + * in a raw dump unless you happen to notice a missing field. So it counts the + * unnamed records explicitly and breaks the rest down by event type, which is + * what turns "there are a ton of save_blobs" into a diagnosis. + */ +async function queueReport (limit = 50): Promise[]> { + const handles = loggersEnabled.filter(l => l.queueDebug).map(l => l.queueDebug!); + if (!handles.length) { + console.log('loDebug: no ack-aware logger with a durable queue.'); + return []; + } + + const rows: Record[] = []; + let total = 0; + let unnamed = 0; + const byType: Record = {}; + + for (const h of handles) { + total += await h.count(); + for (const rec of await h.inspect(limit)) { + // Records are stored as { seq, payload } (memory) or the raw stored + // object (IDB); the payload is the serialized frame either way. + const r = rec as Record; + const raw = (r.payload ?? r) as unknown; + let frame: Record = {}; + try { frame = typeof raw === 'string' ? JSON.parse(raw) : (raw as any) ?? {}; } + catch { /* unparseable — reported as unknown below */ } + + const type = frame.event ?? frame.type ?? '(unknown)'; + const id = frame?.metadata?.eventId; + byType[type] = (byType[type] ?? 0) + 1; + if (!id) unnamed++; + rows.push({ seq: r.seq, event: type, eventId: id ?? '— UNNAMED —', bytes: JSON.stringify(frame).length }); + } + } + + console.log(`loDebug: ${total} record(s) waiting; showing up to ${limit}.`); + console.table(byType); + if (unnamed) { + console.warn( + `loDebug: ${unnamed} of the first ${limit} record(s) inspected have no ` + + 'metadata.eventId (the total above may hold more). The server acks ' + + 'by name, so these can never be acked — they are sent best-effort and ' + + 'dropped. If they keep appearing, an enqueue path is not stamping.' + ); + } + console.table(rows); + return rows; +} + +function clearQueues (): void { + const handles = loggersEnabled.filter(l => l.queueDebug).map(l => l.queueDebug!); + handles.forEach(h => h.clear()); + console.warn(`loDebug: cleared ${handles.length} queue(s) — unsent events discarded.`); +} + +export const loDebug = { queue: queueReport, clearQueue: clearQueues }; + +// Attach for console use. Debug-only affordance, browser-only, and it never +// overwrites something already there. +if (typeof globalThis !== 'undefined' && !(globalThis as any).loDebug) { + (globalThis as any).loDebug = loDebug; +} + // TODO: We should consider specifying a set of verbs, nouns, etc. we // might use, and outlining what can be expected in the protocol // TODO: We should consider structing / destructing here @@ -125,7 +217,6 @@ export function init ( debugLevel = debug.LEVEL.NONE as string, debugDest = [debug.LOG_OUTPUT.CONSOLE] as LogDestination[], useDisabler = true, - queueType = Queue.QueueType.AUTODETECT as string, sendBrowserInfo = false, verboseEvents = false, metadata = [] as MetadataTask[], @@ -135,7 +226,9 @@ export function init ( if (!version || typeof version !== 'string') throw new Error('version must be a non-null string'); util.setVerboseEvents(verboseEvents); - queue = new Queue.Queue('LOEvent', { queueType }); + // The front desk only preserves pre-go ordering. Durability belongs to each + // logger's outbox, so a second disk queue here would add a destructive hop. + queue = new Queue.Queue('LOEvent', { queueType: Queue.QueueType.IN_MEMORY }); debug.setLevel(debugLevel); debug.setLogOutputs(debugDest); @@ -144,6 +237,7 @@ export function init ( } loggersEnabled = loggers; + loggersEnabled.forEach(logger => logger.configure?.({ source })); initialized = INIT_STATES.IN_PROGRESS; pendingSource = source; pendingVersion = version; @@ -183,7 +277,6 @@ export function go () { initialized = INIT_STATES.READY; queue.startDequeueLoop({ initialize: isInitialized, - shouldDequeue: disabler.retry, onDequeue: sendEvent }); }); @@ -196,17 +289,19 @@ function sendEvent (event: unknown) { logger(jsonEncodedEvent); } catch (error) { if (error instanceof disabler.BlockError) { - // Handle BlockError exception here disabler.handleBlockError(error); } else { - // Other types of exceptions will propagate up - throw error; + // One logger must never cost its siblings their copy of an event. + debug.error(`Logger ${logger.lo_id ?? logger.lo_name ?? 'unnamed'} threw on an event`, error); } } } } export function logEvent (eventType: string, event: Record) { + if (util.isProtocolEventName(eventType)) { + throw new Error(`logEvent: '${eventType}' is a reserved protocol frame name`); + } // opt out / dead if (!disabler.storeEvents()) { return; diff --git a/src/memoryQueue.ts b/src/memoryQueue.ts index 43c00fb..7563b42 100644 --- a/src/memoryQueue.ts +++ b/src/memoryQueue.ts @@ -4,44 +4,121 @@ * - Works everywhere / act as a fallback where indexeddb is unavailable * - Nice for dev, where we don't want to persist events from buggy code * - Nice for simple use-cases + * + * It implements two dequeue disciplines (see QueueBackend in types.ts): + * - dequeue(): destructive take (delete-on-read). + * - leaseNext()/confirm()/rewind(): non-destructive lease for the ack + * protocol — an item stays until confirm()ed, and rewind() re-hands + * everything unconfirmed. + * A given instance should use one discipline, not both. */ +import type { LeasedItem } from './types.js'; + +interface Entry { seq: number; payload: unknown; } export class Queue { - private queue: unknown[]; - private queueName: string; - private promise: Promise | null; - private resolve: ((value: unknown) => void) | null; + private items: Entry[]; + private readonly queueName: string; + private nextSeq: number; + // Highest seq handed out by leaseNext() this session. leaseNext() returns + // the lowest stored item with seq > leasedThrough; rewind() resets it so + // unconfirmed items are re-handed. + private leasedThrough: number; + // A single parked consumer (dequeue or leaseNext) waiting on an empty queue. + private waiter: { resolve: (value: unknown) => void; lease: boolean } | null; constructor (queueName: string) { - this.queue = []; + this.items = []; this.queueName = queueName; - this.promise = null; - this.resolve = null; + this.nextSeq = 1; + this.leasedThrough = 0; + this.waiter = null; this.enqueue = this.enqueue.bind(this); this.dequeue = this.dequeue.bind(this); + this.leaseNext = this.leaseNext.bind(this); + this.confirm = this.confirm.bind(this); + this.rewind = this.rewind.bind(this); + this.unconfirmedCount = this.unconfirmedCount.bind(this); } async initialize () { } + async inspect (limit: number): Promise { + return this.items.slice(0, limit).map(e => ({ seq: e.seq, payload: e.payload })); + } + + clear () { + this.items = []; + this.leasedThrough = 0; + } + enqueue (item: unknown) { - if (this.promise) { - this.resolve!(item); - this.promise = null; - } else { - this.queue.push(item); + this.items.push({ seq: this.nextSeq++, payload: item }); + this.wake(); + } + + /** Wake by re-running the queue's normal scan. Direct hand-off can advance a + * cursor past an older record and violates the shared backend contract. */ + private wake () { + const waiter = this.waiter; + if (!waiter) return; + if (waiter.lease) { + const next = this.items.find(entry => entry.seq > this.leasedThrough); + if (!next) return; + this.waiter = null; + this.leasedThrough = next.seq; + waiter.resolve({ seq: next.seq, item: next.payload }); + return; } + if (!this.items.length) return; + this.waiter = null; + waiter.resolve(this.items.shift()!.payload); } dequeue (): unknown | Promise { - if (this.queue.length > 0) { - return this.queue.shift(); - } else { - this.promise = new Promise((resolve) => { - this.resolve = resolve; - }); - return this.promise; + if (this.items.length > 0) { + return (this.items.shift() as Entry).payload; + } + return new Promise((resolve) => { + this.waiter = { resolve, lease: false }; + }); + } + + leaseNext (): Promise { + const next = this.items.find(e => e.seq > this.leasedThrough); + if (next) { + this.leasedThrough = next.seq; + return Promise.resolve({ seq: next.seq, item: next.payload }); } + return new Promise((resolve) => { + this.waiter = { resolve: resolve as (value: unknown) => void, lease: true }; + }); + } + + confirm (seqs: number[]) { + if (!seqs.length) return; + const drop = new Set(seqs); + this.items = this.items.filter(e => !drop.has(e.seq)); + } + + rewind () { + this.leasedThrough = 0; + this.wake(); + } + + unconfirmedCount (): number { + return this.items.length; + } + + /** Highest stored seq (items are kept in ascending seq), or null if empty. */ + async maxSeq (): Promise { + return this.items.length ? this.items[this.items.length - 1].seq : null; + } + + /** Stored records at or below `seq` that this instance has not leased. */ + async unleasedAtOrBelow (seq: number): Promise { + return this.items.filter(e => e.seq > this.leasedThrough && e.seq <= seq).length; } } diff --git a/src/protocol.ts b/src/protocol.ts new file mode 100644 index 0000000..3a1b886 --- /dev/null +++ b/src/protocol.ts @@ -0,0 +1,277 @@ +import type { Decision, DeliveryOptions } from './types.js'; + +export const PROBE_INTERVAL_MS = 300; +export const BARRIER_DEADLINE_MS = 5_000; +export const SNAPSHOT_RETRY_MS = 10_000; + +type Barrier = 'unevaluated' | 'pending' | 'clear'; + +/** + * Reliable-delivery policy with no socket, storage, timer, or ambient clock. + * + * Methods receive facts and return decisions. The websocket adapter performs + * those decisions and reports asynchronous outcomes back with the connection + * generation that produced them. Identity acknowledgements are deliberately + * generation-free: an ack is a durable fact about an event, not a socket. + */ +export class DeliveryEngine { + private readonly autoack: boolean; + private generationNumber = 0; + private online = false; + private paused = false; + private sending = false; + private currentSend: { seq: number; eventId: string | null } | null = null; + private readonly inFlight = new Map(); + + private barrier: Barrier = 'unevaluated'; + private watermark: number | null = null; + private barrierElapsed = 0; + private probeElapsed = 0; + private lastUnleased: number | null = null; + + private snapshotFrame: string | null = null; + private snapshotSending = false; + private snapshotSent = false; + private snapshotElapsed = 0; + + constructor ({ autoack = false }: DeliveryOptions = {}) { + this.autoack = autoack; + } + + generation (): number { return this.generationNumber; } + barrierIsClear (): boolean { return this.barrier === 'clear'; } + awaitingAck (): number { return this.inFlight.size; } + + connected (): Decision[] { + this.generationNumber++; + this.online = true; + this.sending = false; + this.currentSend = null; + this.barrier = 'unevaluated'; + this.watermark = null; + this.barrierElapsed = 0; + this.probeElapsed = 0; + this.lastUnleased = null; + this.snapshotSending = false; + this.snapshotSent = false; + this.snapshotElapsed = 0; + + const decisions: Decision[] = [{ do: 'rewind' }]; + if (!this.paused) decisions.push({ do: 'resumeSending' }); + decisions.push({ do: 'measureWatermark' }); + return decisions; + } + + disconnected (): Decision[] { + this.generationNumber++; + this.online = false; + this.sending = false; + this.currentSend = null; + this.barrier = 'unevaluated'; + this.snapshotSending = false; + this.snapshotSent = false; + this.snapshotElapsed = 0; + return [{ do: 'pauseSending' }]; + } + + recordLeased (seq: number, eventId: string | null, frame: string): Decision[] { + if (!this.online || this.paused) return [{ do: 'rewind' }]; + + this.sending = true; + this.currentSend = { seq, eventId }; + const decisions: Decision[] = []; + if (eventId === null) { + decisions.push({ + do: 'log', + level: 'error', + message: `outbox record ${seq} has no metadata.eventId; sending best-effort and confirming on send` + }); + } else if (!this.autoack) { + // Register before the adapter sends: a fast ack must already have a name + // to storage-id mapping. + this.inFlight.set(eventId, seq); + } + decisions.push({ do: 'sendFrame', frame, seq }); + return decisions; + } + + sendCompleted (generation: number): Decision[] { + if (!this.isCurrent(generation) || !this.sending || !this.currentSend) return []; + const { seq, eventId } = this.currentSend; + this.sending = false; + this.currentSend = null; + + const decisions: Decision[] = []; + if (this.autoack || eventId === null) decisions.push({ do: 'confirmIds', ids: [seq] }); + decisions.push(...this.probeIfPending()); + return decisions; + } + + sendFailed (generation: number): Decision[] { + if (!this.isCurrent(generation) || !this.sending || !this.currentSend) return []; + const { seq, eventId } = this.currentSend; + this.sending = false; + this.currentSend = null; + if (eventId !== null && this.inFlight.get(eventId) === seq) { + this.inFlight.delete(eventId); + } + return [ + { do: 'log', level: 'error', message: `send of outbox record ${seq} failed; rewinding` }, + { do: 'rewind' } + ]; + } + + ackReceived (eventId: string): Decision[] { + if (this.autoack) return []; + const seq = this.inFlight.get(eventId); + if (seq === undefined) return []; + this.inFlight.delete(eventId); + return [{ do: 'confirmIds', ids: [seq] }, ...this.probeIfPending()]; + } + + watermarkResult (generation: number, maxSeq: number | null): Decision[] { + if (!this.isCurrent(generation) || this.barrier !== 'unevaluated') return []; + if (maxSeq === null) { + this.barrier = 'clear'; + return this.askIfReady(); + } + this.watermark = maxSeq; + this.barrier = 'pending'; + this.barrierElapsed = 0; + this.probeElapsed = 0; + this.lastUnleased = null; + return [{ do: 'probeQueue', watermark: maxSeq }]; + } + + probeResult (generation: number, unleased: number): Decision[] { + if (!this.isCurrent(generation) || this.barrier !== 'pending' || this.sending) return []; + if (unleased > 0) { + if (this.lastUnleased !== null && unleased < this.lastUnleased) { + this.barrierElapsed = 0; + } + this.lastUnleased = unleased; + return []; + } + this.barrier = 'clear'; + return this.askIfReady(); + } + + measurementFailed (generation: number, operation: string): Decision[] { + if (!this.isCurrent(generation) || this.barrier === 'clear') return []; + this.barrier = 'clear'; + return [ + { + do: 'log', + level: 'error', + message: `flush barrier ${operation} failed; requesting potentially stale state` + }, + ...this.askIfReady() + ]; + } + + requestState (frame: string): Decision[] { + this.snapshotFrame = frame; + this.snapshotSending = false; + this.snapshotSent = false; + this.snapshotElapsed = 0; + return this.askIfReady(); + } + + /** The direct request reached a verified-open socket. Only now does its + * retry clock begin; merely deciding to ask is not a successful send. */ + stateSendCompleted (generation: number): Decision[] { + if (!this.isCurrent(generation) || !this.snapshotSending || !this.snapshotFrame) return []; + this.snapshotSending = false; + this.snapshotSent = true; + this.snapshotElapsed = 0; + return []; + } + + stateSendFailed (generation: number): Decision[] { + if (!this.isCurrent(generation) || !this.snapshotSending) return []; + this.snapshotSending = false; + return [{ do: 'log', level: 'error', message: 'state snapshot request did not reach an open socket; reconnecting before retry' }]; + } + + stateReceived (): Decision[] { + this.snapshotFrame = null; + this.snapshotSending = false; + this.snapshotSent = false; + this.snapshotElapsed = 0; + return []; + } + + disablerEngaged ({ permanentOptOut = false, permanent = false } = {}): Decision[] { + this.paused = true; + const decisions: Decision[] = [{ do: 'pauseSending' }]; + if (permanentOptOut) { + decisions.push( + { do: 'log', level: 'error', message: 'permanent privacy opt-out: discarding the stored outbox' }, + { do: 'discardOutbox' } + ); + } else if (permanent) { + decisions.push({ do: 'log', level: 'error', message: 'delivery is permanently paused; retained events will remain in the outbox' }); + } + return decisions; + } + + disablerReleased (): Decision[] { + if (!this.paused) return []; + this.paused = false; + return this.online ? [{ do: 'resumeSending' }] : []; + } + + elapsed (ms: number): Decision[] { + if (!this.online) return []; + const decisions: Decision[] = []; + + // The deadline covers *unevaluated* as well as pending. A maxSeq promise + // that never settles must degrade to stale state, never a permanent spinner. + if (this.barrier !== 'clear') { + this.barrierElapsed += ms; + if (this.barrierElapsed >= BARRIER_DEADLINE_MS) { + this.barrier = 'clear'; + decisions.push({ + do: 'log', + level: 'error', + message: 'flush barrier did not settle before its deadline; requesting potentially stale state' + }); + decisions.push(...this.askIfReady()); + } else if (this.barrier === 'pending') { + this.probeElapsed += ms; + if (this.probeElapsed >= PROBE_INTERVAL_MS && !this.sending) { + this.probeElapsed = 0; + decisions.push({ do: 'probeQueue', watermark: this.watermark! }); + } + } + } + + if (this.snapshotFrame && this.snapshotSent) { + this.snapshotElapsed += ms; + if (this.snapshotElapsed >= SNAPSHOT_RETRY_MS) { + this.snapshotSent = false; + this.snapshotElapsed = 0; + decisions.push({ do: 'log', level: 'error', message: 'state snapshot request timed out; asking again' }); + decisions.push(...this.askIfReady()); + } + } + return decisions; + } + + private isCurrent (generation: number): boolean { + return this.online && generation === this.generationNumber; + } + + private probeIfPending (): Decision[] { + if (this.barrier !== 'pending' || this.sending) return []; + this.probeElapsed = 0; + return [{ do: 'probeQueue', watermark: this.watermark! }]; + } + + private askIfReady (): Decision[] { + if (!this.online || this.barrier !== 'clear' || !this.snapshotFrame) return []; + if (this.snapshotSending || this.snapshotSent) return []; + this.snapshotSending = true; + return [{ do: 'askForState', frame: this.snapshotFrame }]; + } +} diff --git a/src/queue.ts b/src/queue.ts index a49896f..4f7c564 100644 --- a/src/queue.ts +++ b/src/queue.ts @@ -2,7 +2,7 @@ import * as indexeddbQueue from './indexeddbQueue.js'; import * as memoryQueue from './memoryQueue.js'; import * as debug from './debugLog.js'; import * as util from './util.js'; -import type { QueueBackend, DequeueLoopConfig } from './types.js'; +import type { QueueBackend, DequeueLoopConfig, LeasedItem } from './types.js'; export const QueueType = { AUTODETECT: 'AUTODETECT', // Persistent if available, otherwise in-memory @@ -24,21 +24,16 @@ function autodetect () { } export class Queue { - private queue: QueueBackend; + private readonly queue: QueueBackend; startDequeueLoop: (config: DequeueLoopConfig) => Promise; constructor (queueName: string, { queueType = QueueType.AUTODETECT as string } = {}) { - if (queueType === QueueType.AUTODETECT) { - queueType = autodetect(); - } + if (queueType === QueueType.AUTODETECT) queueType = autodetect(); const QueueClass = queueClasses[queueType]; - if (QueueClass) { - debug.info(`Queue: using ${queueType.toLowerCase()}Queue`); - this.queue = new QueueClass(queueName); - } else { - throw new Error('Invalid queue type'); - } + if (!QueueClass) throw new Error(`Invalid queue type: ${queueType}`); + debug.info(`Queue: ${queueName} using ${queueType.toLowerCase()}Queue`); + this.queue = new QueueClass(queueName); this.enqueue = this.enqueue.bind(this); this.startDequeueLoop = util.once(this._startDequeueLoop.bind(this)); @@ -48,6 +43,46 @@ export class Queue { this.queue.enqueue(item); } + leaseNext (): Promise { return this.queue.leaseNext(); } + + /** Delete exactly the listed seqs — the ones THIS connection sent and saw + * acked. Never a range: the store is shared across tabs. */ + confirm (seqs: number[]) { + this.queue.confirm(seqs); + } + + /** Reset the lease cursor so unconfirmed items are re-handed (resend). */ + rewind () { + this.queue.rewind(); + } + + /** Count of enqueued-but-unconfirmed items (drives the unsaved warning). */ + unconfirmedCount (): Promise | number { + return this.queue.unconfirmedCount(); + } + + /** Highest stored seq, or null when empty — the snapshot barrier watermark, + * captured once per connection after rewind. */ + maxSeq (): Promise { + return this.queue.maxSeq(); + } + + /** Stored records at or below `seq` that this instance has not yet leased. + * Zero means the flush barrier is clear (see QueueBackend in types.ts). */ + unleasedAtOrBelow (seq: number): Promise { + return this.queue.unleasedAtOrBelow(seq); + } + + /** DEBUG: peek at what is sitting in the queue. */ + inspect (limit = 20): Promise { + return this.queue.inspect(limit); + } + + /** DEBUG / RECOVERY: drop everything, unsent included. */ + clear () { + this.queue.clear(); + } + /** * This function starts a loop to continually * dequeue items and process them appropriately @@ -55,10 +90,13 @@ export class Queue { */ private async _startDequeueLoop ({ initialize = async () => true, - shouldDequeue = async () => true, onDequeue = async (_item: unknown) => {}, onError = (message: string, error: unknown) => debug.error(message, error) }: DequeueLoopConfig = {}) { + if (!this.queue.dequeue) { + onError('QUEUE ERROR: this backend has no destructive dequeue', new Error('lease-only backend')); + return; + } try { if (!await initialize()) { throw new Error('QUEUE ERROR: Initialization function returned false.'); @@ -70,24 +108,6 @@ export class Queue { debug.info('QUEUE: Dequeue loop initialized.'); while (true) { - // Check if we are allowed to continue dequeueing. - // When shouldDequeue() returns false, we permanently terminate - // the loop. This is intentional — the primary caller is - // disabler.retry(), which only returns false for permanent - // opt-outs (e.g. student privacy requests). In that case, - // the loop must stop and must not restart. Temporary blocks - // (e.g. rate limits) are handled inside disabler.retry() by - // awaiting the expiration before returning true. - try { - if (!await shouldDequeue()) { - throw new Error('QUEUE ERROR: Dequeue streaming returned false.'); - } - } catch (error) { - onError('QUEUE ERROR: Not allowed to start dequeueing', error); - return; - } - - // do something with the item const item = await this.queue.dequeue(); try { if (item !== null) { diff --git a/src/reduxLogger.ts b/src/reduxLogger.ts index 780ebf6..71295eb 100644 --- a/src/reduxLogger.ts +++ b/src/reduxLogger.ts @@ -32,17 +32,101 @@ declare global { } } +// ============================================================================= +// Types +// ============================================================================= + interface ReduxAction extends JSONObject { redux_type: string; type: string; payload: JSONValue; } +export type SaveStatus = 'saved' | 'modified' | 'error'; + +/** + * Options for the Redux logger's persistence behavior. + * + * serializeForSave: Called before every save (server and localStorage). + * Receives the full Redux state, returns the subset to persist. + * Default: identity (persist everything). + * + * deserializeOnLoad: Called when a fetch_blob response arrives. + * Receives the raw blob from the server and the current Redux state. + * Returns the blob to merge (via shallow spread) into current state. + * Default: identity (merge entire blob). + */ +export interface ReduxLoggerOptions { + serializeForSave?: (state: JSONObject) => JSONObject; + deserializeOnLoad?: (blob: JSONObject, currentState: JSONObject) => JSONObject; + /** + * Cross-tab state sync via redux-state-sync. Default: false (off). + * + * - false (default): nothing is broadcast. + * - true: broadcast every action to other store instances in the same + * browser — EXCEPT lo_event's own lifecycle actions (see below). + * - { predicate }: broadcast only actions `predicate(action)` approves + * (still minus the lifecycle actions). Lets the app drop events that must + * not cross tabs — e.g. content-load events, which are per-tab. + * + * Regardless of true/predicate, lo_event NEVER broadcasts its own lifecycle + * actions (SET_STATE — a full-state replace from blob restore — and + * LOCKFIELDS), since those would clobber or duplicate state across tabs. + * + * Off by default because, unfiltered, lifecycle/content actions (and + * non-idempotent reactive effects) echo between tabs and corrupt state. + */ + stateSync?: boolean | { predicate?: (action: ReduxAction) => boolean }; +} + +// NOTE: a true local-only mode (no fetch_blob server) is not yet supported. +// It needs a real localStorage RESTORE path (there is only a write path today; +// see loadState, commented out below), plus local save-status handling +// (markSaved after the local write instead of waiting for a server ack that +// never comes). Until that feature lands, reduxLogger expects a load cycle +// (fetch_blob) to flip IS_LOADED. + +// ============================================================================= +// Module state +// ============================================================================= + const EMIT_EVENT = 'EMIT_EVENT'; const EMIT_LOCKFIELDS = 'EMIT_LOCKFIELDS'; const EMIT_SET_STATE = 'SET_STATE'; let IS_LOADED = false; +let _options: ReduxLoggerOptions = {}; + +// Cross-tab state sync gating (see ReduxLoggerOptions.stateSync). +// The middleware's predicate reads _stateSyncEnabled at dispatch time, so +// toggling this after the store is created takes effect immediately. The +// incoming-message listener is attached lazily and only when enabled, so a +// disabled store neither sends nor receives. Default off (opt-in). +let _stateSyncEnabled = false; +let _stateSyncListenerAttached = false; +// Optional app-supplied filter for which actions to broadcast (see +// ReduxLoggerOptions.stateSync). null = broadcast all (minus lifecycle). +let _stateSyncPredicate: ((action: ReduxAction) => boolean) | null = null; + +// Whether to broadcast a given action to other tabs. lo_event NEVER broadcasts +// its own lifecycle actions: SET_STATE (full-state replace from blob restore) +// and LOCKFIELDS would clobber/duplicate state across tabs. The app predicate +// filters the rest (e.g. dropping per-tab content-load events). +function shouldBroadcast (action: ReduxAction): boolean { + if (!_stateSyncEnabled) return false; + if (action.redux_type === EMIT_SET_STATE || action.redux_type === EMIT_LOCKFIELDS) return false; + return _stateSyncPredicate ? _stateSyncPredicate(action) : true; +} + +function ensureStateSyncListener () { + // Browser-only: initMessageListener uses the BroadcastChannel, which is + // not available (and not meaningful) server-side. + if (typeof window === 'undefined' || typeof BroadcastChannel === 'undefined') return; + if (_stateSyncEnabled && !_stateSyncListenerAttached) { + initMessageListener(store); + _stateSyncListenerAttached = true; + } +} // TODO: Import debugLog and use those functions. const DEBUG = false; @@ -53,35 +137,102 @@ function debug_log (...args: unknown[]) { } } +// ============================================================================= +// Persistence status — plain external store (NOT in Redux) +// +// These are metadata about the save machinery, not application state. +// Keeping them outside Redux avoids cross-tab dispatch loops via +// redux-state-sync and keeps the store subscription read-only. +// +// React consumers use useSyncExternalStore via the hooks in hooks.ts. +// ============================================================================= + +let _saveStatus: SaveStatus = 'saved'; +let _connected: boolean | null = null; // null = no websocket configured + +// Monotonic token: incremented on each save_blob dispatch, compared against +// the token echoed back in save_blob_ack. Status is 'saved' only when the +// server has confirmed the most recent save. +let _saveToken = 0; +let _ackedToken = 0; + +const _statusListeners = new Set<() => void>(); + +function notifyStatusListeners () { + _statusListeners.forEach(fn => fn()); +} + +function markModified () { + if (_saveStatus !== 'modified') { + _saveStatus = 'modified'; + notifyStatusListeners(); + } +} + +function markSaved () { + if (_saveStatus !== 'saved') { + _saveStatus = 'saved'; + notifyStatusListeners(); + } +} + +// The server reported a save failure. Distinct from 'modified' so the UI can +// tell "still saving" from "save failed". A subsequent change re-enters +// 'modified' (markModified), and the next successful ack clears it to 'saved'. +function markError () { + if (_saveStatus !== 'error') { + _saveStatus = 'error'; + notifyStatusListeners(); + } +} + +function setConnected (value: boolean) { + if (_connected !== value) { + _connected = value; + notifyStatusListeners(); + } +} + +/** Subscribe to persistence status changes (save status, connected, loaded). */ +export function subscribeStatus (listener: () => void): () => void { + _statusListeners.add(listener); + return () => { _statusListeners.delete(listener); }; +} + +/** Snapshot of save status for useSyncExternalStore. */ +export function getSaveStatus (): SaveStatus { return _saveStatus; } + +/** Snapshot of connection status. null = no websocket, true/false = connected/disconnected. */ +export function getConnected (): boolean | null { return _connected; } + +/** Snapshot of loaded status (a fetch_blob load cycle has resolved). */ +export function getLoaded (): boolean { return IS_LOADED; } + +// ============================================================================= +// Load / Save +// ============================================================================= + /** * Update the redux logger's state with `data`. - * This is fired when consuming a custom `fetch_blob` - * event. + * This is fired when consuming a custom `fetch_blob` event. */ export function handleLoadState (data: unknown) { IS_LOADED = true; const state = store.getState() as JSONObject; if (data) { - setState( - { - ...state, - ...data, - settings: { - ...(state.settings as JSONObject), - reduxStoreStatus: IS_LOADED - } - }); + const blob = _options.deserializeOnLoad + ? _options.deserializeOnLoad(data as JSONObject, state) + : data as JSONObject; + setState({ ...state, ...blob }); } else { debug_log('No data provided while handling state from server, continuing.'); - setState( - { - ...state, - settings: { - ...(state.settings as JSONObject), - reduxStoreStatus: IS_LOADED - } - }); } + // After loading, state matches the server — reset tokens and mark saved. + // markSaved() AFTER setState so the subscription's markModified() fires + // first (synchronously from the dispatch), then we correct it here. + _ackedToken = _saveToken; + markSaved(); + notifyStatusListeners(); // loaded changed } async function saveStateToLocalStorage (state: JSONObject) { @@ -92,7 +243,8 @@ async function saveStateToLocalStorage (state: JSONObject) { try { const KEY = (state?.settings as JSONObject)?.reduxID as string || 'redux'; - const serializedState = JSON.stringify(state); + const toSave = _options.serializeForSave ? _options.serializeForSave(state) : state; + const serializedState = JSON.stringify(toSave); localStorage.setItem(KEY, serializedState); } catch (e) { // Ignore @@ -100,8 +252,7 @@ async function saveStateToLocalStorage (state: JSONObject) { } /** - * Dispatch a `save_blob` event on the redux - * logger. + * Dispatch a `save_blob` event on the redux logger. */ async function saveStateToServer (state: JSONObject) { if (!IS_LOADED) { @@ -110,15 +261,28 @@ async function saveStateToServer (state: JSONObject) { } try { - // console.log("dispatching save_blob") - util.dispatchCustomEvent('save_blob', { detail: state }); - // store.dispatch('save_blob', { detail: state }); + const toSave = _options.serializeForSave ? _options.serializeForSave(state) : state; + _saveToken++; + util.dispatchCustomEvent('save_blob', { detail: { blob: toSave, token: _saveToken } }); + // Don't markSaved() here — wait for save_blob_ack from the server. } catch (e) { - // Ignore debug_log('Error in dispatch', { e }); } } +/** + * Immediately flush any pending debounced saves. + * Available for programmatic use (e.g. beforeunload handlers). + */ +export function saveNow () { + debouncedSaveStateToLocalStorage.flush(); + debouncedSaveStateToServer.flush(); +} + +// ============================================================================= +// Action creators +// ============================================================================= + // Action creator function This is a little bit messy, since we // duplicate type from the payload. It's not clear if this is a good // idea. We used to have `type` be set to the current contents of @@ -150,6 +314,10 @@ const emitSetState = (state: JSONObject): ReduxAction => { }; }; +// ============================================================================= +// Reducers +// ============================================================================= + function store_last_event_reducer (state: JSONObject = {}, action: JSONObject): JSONObject { const a = action as ReduxAction; return { ...state, event: a.payload }; @@ -271,6 +439,10 @@ const reducer = (state: JSONObject = {}, action: ReduxAction): JSONObject => { return state; }; +// ============================================================================= +// Store +// ============================================================================= + const eventQueue: unknown[] = []; const composeEnhancers = (typeof window !== 'undefined' && window.__REDUX_DEVTOOLS_EXTENSION_COMPOSE__) || redux.compose; @@ -281,13 +453,31 @@ const composeEnhancers = (typeof window !== 'undefined' && window.__REDUX_DEVTOO // back to thunk. // const presistedState = loadState(); +// Cross-tab sync is a browser concept and createStateSyncMiddleware() +// constructs a BroadcastChannel eagerly. In Node (server-side, SSR) that +// hits broadcast-channel's filesystem fallback and throws, so we only add +// the middleware in a browser. Note: a config object MUST include `channel` +// — redux-state-sync replaces its whole defaultConfig with the passed config, +// so omitting it yields `new BroadcastChannel(undefined)` and crashes. +const _baseMiddleware: redux.Middleware[] = [((thunk as any).default || thunk) as redux.Middleware]; +if (typeof window !== 'undefined' && typeof BroadcastChannel !== 'undefined') { + // predicate gates outgoing broadcasts; the incoming listener is attached + // separately in ensureStateSyncListener(). Both respect _stateSyncEnabled. + _baseMiddleware.push(createStateSyncMiddleware({ + channel: 'redux_state_sync', + predicate: (action: any) => shouldBroadcast(action as ReduxAction), + }) as redux.Middleware); +} + export let store: redux.Store> = redux.createStore( reducer as unknown as redux.Reducer>, { event: null } as unknown as JSONObject, // Base state - composeEnhancers(redux.applyMiddleware(((thunk as any).default || thunk) as redux.Middleware, createStateSyncMiddleware() as redux.Middleware)) + composeEnhancers(redux.applyMiddleware(..._baseMiddleware)) ); -initMessageListener(store); +// initMessageListener is attached lazily by reduxLogger() when stateSync is +// enabled — see ensureStateSyncListener(). Attaching it unconditionally here +// would make a disabled store still receive (and respond to) broadcasts. let promise: (Promise & { resolve?: (value: unknown) => void }) | null = null; let previousEventString: string | null = null; @@ -316,13 +506,8 @@ function composeReducers(...reducers: ReducerFn[]): ReducerFn { export function setState(state: JSONObject) { debug_log('Set state called'); if (Object.keys(state).length === 0) { - const storeState = store.getState() as JSONObject; - state = { - settings: { - ...(storeState.settings as JSONObject), - reduxStoreStatus: IS_LOADED - } - }; + debug_log('setState called with empty object — ignoring'); + return; } store.dispatch(emitSetState(state) as unknown as redux.Action); } @@ -335,11 +520,18 @@ const debouncedSaveStateToServer = debounce((state: JSONObject) => { saveStateToServer(state); }, 1000); +// ============================================================================= +// Store initialization & subscription +// ============================================================================= + function initializeStore () { + // The subscription is read-only — it never dispatches to the store. + // Save status lives in a plain module-level variable (see above), + // avoiding cross-tab loops via redux-state-sync. store.subscribe(() => { const state = store.getState() as JSONObject; - // we use debounce to save the state once every second - // for better performances in case multiple changes occur in a short time + + markModified(); debouncedSaveStateToLocalStorage(state); debouncedSaveStateToServer(state); @@ -364,16 +556,41 @@ function initializeStore () { // to have this behind a flag later. eventQueue.push(event); } - for (const i in eventSubscribers) { - eventSubscribers[i](event); + for (const subscriber of eventSubscribers) { + subscriber(event); } }); + + // Flush any pending saves when the page is about to close. + // The IndexedDB-backed queue in websocketLogger survives page close, + // so even if the browser terminates before the WebSocket send completes, + // the blob is persisted locally and transmitted on the next page load. + if (typeof window !== 'undefined') { + window.addEventListener('beforeunload', () => { + saveNow(); + }); + } } -export function reduxLogger (subscribers?: Array<(event: unknown) => void>, initialState: JSONObject | null = null): Logger { +// ============================================================================= +// Logger factory +// ============================================================================= + +export function reduxLogger (subscribers?: Array<(event: unknown) => void>, options: ReduxLoggerOptions = {}): Logger { if (subscribers != null) { eventSubscribers = subscribers; } + _options = options; + + // Opt-in (default false). `true` or a predicate enables it; a predicate also + // filters which actions broadcast (see shouldBroadcast). When enabled, attach + // the incoming listener (once); when disabled, the predicate stops all + // outgoing broadcasts and we never attach the listener, so nothing is received. + _stateSyncEnabled = options.stateSync != null && options.stateSync !== false; + _stateSyncPredicate = (typeof options.stateSync === 'object' && options.stateSync !== null) + ? (options.stateSync.predicate ?? null) + : null; + ensureStateSyncListener(); const logEvent: Logger = function (event: string) { store.dispatch(emitEvent(event) as unknown as redux.Action); @@ -391,10 +608,6 @@ export function reduxLogger (subscribers?: Array<(event: unknown) => void>, init logEvent.getLockFields = function () { return lockFields; }; - // do we want to initialize the store here? We set it to the stored state in create store - // if (initialState) { - // } - return logEvent; } @@ -446,6 +659,35 @@ export function handleAuth (user: unknown) { })) as unknown as redux.Action); } -// Start listening for fetch +// ============================================================================= +// CustomEvent listeners +// ============================================================================= + util.consumeCustomEvent('fetch_blob', handleLoadState); util.consumeCustomEvent('auth', handleAuth); + +// Connection status from websocketLogger +util.consumeCustomEvent('lo_connection_status', (data: unknown) => { + const { connected } = data as { connected: boolean }; + setConnected(connected); +}); + +// Server acknowledgment of a save_blob write. +// Only mark saved if this ack is for the most recent save — stale acks +// (from earlier saves) are ignored because a newer save is still pending. +util.consumeCustomEvent('save_blob_ack', (data: unknown) => { + const { token } = data as { token: number }; + if (token > _ackedToken) { + _ackedToken = token; + } + if (_ackedToken >= _saveToken) { + markSaved(); + } +}); + +// Server reported a save_blob write failure. Don't advance _ackedToken — the +// blob did not persist — and surface the failure so the UI isn't stuck looking +// like a save is merely in progress. +util.consumeCustomEvent('save_blob_nack', () => { + markError(); +}); diff --git a/src/types.ts b/src/types.ts index 94b984a..ea9425b 100644 --- a/src/types.ts +++ b/src/types.ts @@ -36,12 +36,41 @@ export type ReducerFn = (state: JSONObject, action: JSONObject) => JSONObject; export interface Logger { (event: string): void; init?: () => Promise | void; + /** Called by lo_event.init before init(). Lets a logger derive stable storage + * names from application identity without forcing every caller to repeat it. */ + configure?: (context: { source: string }) => void; setField?: (data: string) => void; lo_name?: string; lo_id?: string; getLockFields?: () => Record | null; + /** Enqueued-but-unacked count (ack-aware loggers, e.g. websocketLogger). */ + unackedCount?: () => Promise | number; + /** Ask (or re-ask) for the current state snapshot. */ + requestState?: () => void; + /** Console debug handles for this logger's durable queue, if it has one. */ + queueDebug?: QueueDebug; } +/** The application chooses one delivery profile; it is never negotiated. */ +export interface DeliveryOptions { + /** false (default): confirm only on server ack. true: confirm after a send + * accepted by a verified-open socket. */ + autoack?: boolean; +} + +/** Decisions emitted by the sans-I/O delivery engine. */ +export type Decision = + | { do: 'rewind' } + | { do: 'measureWatermark' } + | { do: 'probeQueue'; watermark: number } + | { do: 'sendFrame'; frame: string; seq: number } + | { do: 'confirmIds'; ids: number[] } + | { do: 'askForState'; frame: string } + | { do: 'pauseSending' } + | { do: 'resumeSending' } + | { do: 'discardOutbox' } + | { do: 'log'; level: 'info' | 'error'; message: string }; + /** * Metadata task descriptor — used in compileMetadata. * Each task has a name and an async function that produces a result. @@ -51,21 +80,89 @@ export interface MetadataTask { func: () => unknown | Promise; } +/** + * A queued item paired with its durable sequence number, handed out by + * leaseNext(). The seq is assigned at enqueue time, is monotonic per queue, + * and survives reloads (IndexedDB autoIncrement id; a persisted counter in + * memory). It is what the ack protocol confirms — see confirm(). + */ +export interface LeasedItem { + seq: number; + item: unknown; +} + /** * Queue backend interface — the contract both memoryQueue and * indexeddbQueue implement. + * + * Two dequeue disciplines coexist: + * - dequeue(): destructive take (delete-on-read). Used by the front-desk + * loop and simple consumers that don't need delivery proof. + * - leaseNext()/confirm()/rewind(): non-destructive lease. An item stays in + * durable storage until confirm() acks it; rewind() re-hands + * everything unconfirmed (resend on reconnect). This is the + * ack protocol's backbone — nothing is deleted until the + * recipient signs for it. + * A given Queue instance uses ONE discipline; mixing them on one instance is + * unsupported. */ +/** Console-facing handles for a durable queue (see loEvent.queueDebug). */ +export interface QueueDebug { + count(): Promise | number; + inspect(limit?: number): Promise; + clear(): void; +} + export interface QueueBackend { enqueue(item: unknown): void; - dequeue(): unknown | Promise; + dequeue?(): unknown | Promise; + /** Next un-leased stored item (lowest seq), WITHOUT deleting it. Parks + * until an item is available. Advances an in-memory lease cursor. */ + leaseNext(): Promise; + /** Delete EXACTLY the listed stored seqs. + * + * Deliberately not a range delete. The store is shared by every tab in the + * browser (that is what gives tab-close recovery), while each tab acks over + * its OWN socket. A cumulative range delete therefore let one tab's ack + * delete another tab's records — including records that had been enqueued + * but never sent by anyone, which is data loss, not a duplicate. A caller + * passes only the seqs it sent and saw acked on its own connection. */ + confirm(seqs: number[]): void; + /** Reset the lease cursor so leaseNext() re-hands all unconfirmed items + * from the lowest stored seq (used on reconnect to resend). */ + rewind(): void; + /** Count of stored (enqueued, not yet confirmed) items. */ + unconfirmedCount(): Promise | number; + /** Highest stored seq, or null when the store is empty. Captured once per + * connection (after rewind) as the snapshot barrier's watermark: everything + * at or below it is "the backlog this connection started with". */ + maxSeq(): Promise; + /** Stored records with seq at or below `seq` that THIS instance has not yet + * leased (seq > lease cursor). This is the barrier question itself, asked + * of the queue because only the queue knows: on a shared store, another + * tab can send-and-delete records this instance was never going to see, so + * no captured count or send tally can answer it. Zero = barrier clear — + * everything below the watermark was either sent by this connection or + * deleted by an ack (meaning the server already has it). */ + unleasedAtOrBelow(seq: number): Promise; + /** DEBUG: the first `limit` stored items, without leasing or deleting. + * For answering "what is stuck in there, and why?" from a console. */ + inspect(limit: number): Promise; + /** DEBUG / RECOVERY: drop everything, unsent included. Destructive and + * deliberately so — the use case is a queue holding junk (e.g. frames a + * broken build could never get acked) that would otherwise be resent on + * every reconnect forever. */ + clear(): void; } /** * Configuration for the dequeue loop in queue.ts. + * + * This loop belongs only to loEvent's in-memory front desk. The websocket + * adapter owns the outbox lease loop beside the socket it controls. */ export interface DequeueLoopConfig { initialize?: () => Promise | boolean; - shouldDequeue?: () => Promise | boolean; onDequeue?: (item: unknown) => Promise | void; onError?: (message: string, error: unknown) => void; } @@ -85,7 +182,6 @@ export interface InitOptions { debugLevel?: string; debugDest?: unknown[]; useDisabler?: boolean; - queueType?: string; sendBrowserInfo?: boolean; verboseEvents?: boolean; metadata?: MetadataTask[]; diff --git a/src/util.ts b/src/util.ts index b381f93..c4aa1aa 100644 --- a/src/util.ts +++ b/src/util.ts @@ -3,6 +3,13 @@ import { v4 as uuidv4 } from 'uuid'; import { storage } from './browserStorage.js'; +/** Application events share their `event` field with protocol frames. */ +const RESERVED_EVENT_NAMES = new Set(['fetch_blob', 'save_blob', 'lock_fields']); + +export function isProtocolEventName (eventName: string): boolean { + return RESERVED_EVENT_NAMES.has(eventName); +} + /** * Helper function for copying specific field values * from a given source. This is called to collect browser @@ -116,7 +123,8 @@ export function setVerboseEvents(value: boolean): void { * event = { event: 'ADD', data: 'stuff' } * timestampEvent(event) * event - * // { event: 'ADD', data: 'stuff', metadata: { ts, human_ts, iso_ts, sessionIndex, sessionTag } } + * // { event: 'ADD', data: 'stuff', metadata: { ts, human_ts, iso_ts, eventId, + * // browserTag, sessionTag, sessionSeq } } */ export function timestampEvent (event: Record): void { if (!event.metadata) { @@ -125,12 +133,37 @@ export function timestampEvent (event: Record): void { const metadata = event.metadata as Record; metadata.iso_ts = new Date().toISOString(); + + // IDENTITY — always stamped, never gated on verboseEvents. + // + // `eventId` is `..`, and it is the name the ack + // protocol references, so it is load-bearing rather than a debugging + // nicety. Three properties earn it that job: + // - it means the same thing to everyone, forever (unlike a per-connection + // counter, which is meaningful only to the socket that issued it), so an + // ack is a fact about the world: "the server durably has this event"; + // - any tab can therefore act on an ack for a record it did not send — + // which is what lets one tab drain another's leftovers safely; + // - it survives reconnects, so a server can eventually say "I already have + // through " and skip a resend. + // + // `session` is one JS CONTEXT's lifetime — a page load, an extension + // background page, a worker, a node process. Deliberately not "tab": lo_event + // runs where there is no tab, and a name that is false in a real deployment + // is worse than a slightly abstract one. + const seq = eventIndex++; + metadata.browserTag = browserStamp(); + metadata.sessionTag = sessionStamp; + metadata.sessionSeq = seq; + // OPAQUE: joined with "." purely for legibility. The parts are themselves + // uuid-timestamp strings containing "-" (and could gain more), so this is + // NOT a parseable encoding — compare it and grep it, never split it apart. + // The components are alongside for anything that needs them structurally. + metadata.eventId = `${metadata.browserTag as string}.${sessionStamp}.${seq}`; + if(verboseEvents) { metadata.ts = Date.now(); metadata.human_ts = Date(); - metadata.sessionIndex = eventIndex++; - metadata.sessionTag = sessionStamp; - metadata.browserTag = browserStamp(); } } @@ -238,6 +271,41 @@ export async function mergeMetadata (inputList: MetadataInput[]): Promise { return new Promise(resolve => setTimeout(resolve, ms)); } + +// Persistent failure log (the localStorage leg of the failure heuristic: +// console + localStorage + consumer surface; see README "Failure handling"). +// +// Budget by calculation, not vibes: browsers guarantee ~5 MB per origin for +// localStorage, and lo_event already stores the redux blob there. We cap the +// failure log at a small, fixed slice and ring-buffer it (drop oldest lines), +// so it can never grow toward the quota no matter how many failures occur. +// 128 K chars ≈ 256 KB (UTF-16) ≈ ~5% of the guaranteed budget — hundreds of +// entries, leaving the rest for the app and the blob. +const FAILURE_LOG_KEY = 'lo_event_failures'; +const FAILURE_LOG_MAX_CHARS = 128 * 1024; + +/** + * Append a failure record to a bounded, ring-buffered localStorage log + * (NDJSON, newest last). Deliberately NOT wired to every debug.error — only + * call it for notable failures worth persisting, so the log can't grow + * exponentially. No-op (console remains the record) outside a browser or if + * storage is unavailable/full. + */ +export function recordFailure (entry: Record): void { + if (typeof localStorage === 'undefined') return; + try { + const line = JSON.stringify({ ...entry, ts: new Date().toISOString() }); + const prev = localStorage.getItem(FAILURE_LOG_KEY) || ''; + let next = prev ? `${prev}\n${line}` : line; + // Ring-buffer: drop whole oldest lines until within budget. + while (next.length > FAILURE_LOG_MAX_CHARS && next.includes('\n')) { + next = next.slice(next.indexOf('\n') + 1); + } + localStorage.setItem(FAILURE_LOG_KEY, next); + } catch { + // Quota exceeded or storage blocked — console already has it; drop. + } +} const MS = 1; const SECS = 1000 * MS; const MINS = 60 * SECS; diff --git a/src/websocketLogger.ts b/src/websocketLogger.ts index 617a382..3f994dd 100644 --- a/src/websocketLogger.ts +++ b/src/websocketLogger.ts @@ -1,9 +1,10 @@ -import { Queue } from './queue.js'; +import { Queue, QueueType } from './queue.js'; +import { DeliveryEngine } from './protocol.js'; import * as disabler from './disabler.js'; import * as util from './util.js'; import * as debug from './debugLog.js'; import { storage } from './browserStorage.js'; -import type { Logger } from './types.js'; +import type { Decision, Logger } from './types.js'; interface WsHostOverrides { hostname?: string; @@ -12,217 +13,494 @@ interface WsHostOverrides { url?: string; } -function wsHost(overrides: WsHostOverrides = {}, loc = window.location) { - const { hostname, port, path, url } = overrides; +export interface WebsocketLoggerOptions { + /** false (default): retain until server ack. true: confirm on OPEN send. */ + autoack?: boolean; + /** Stable application namespace. lo_event.init's source is used by default. */ + namespace?: string; + /** Whether init should request a state snapshot. Defaults to !autoack. */ + fetchState?: boolean; + /** Outbox backend; primarily useful for tests and non-browser runtimes. */ + queueType?: string; +} + +type SocketConstructor = new (url: string) => WebSocket; +const SOCKET_OPEN = 1; +const TICK_MS = 250; +const RETRY_PAUSE_MS = 50; +const FETCH_BLOB_FRAME = JSON.stringify({ event: 'fetch_blob' }); + +function defaultLocation (): Location { + if (typeof window === 'undefined') throw new Error('A websocket URL is required outside a browser.'); + return window.location; +} + +function wsHost (overrides: WsHostOverrides = {}, loc = defaultLocation()): string { const protocol = loc.protocol === 'https:' ? 'wss://' : 'ws://'; - const host = hostname || loc.hostname; - const portNumber = port || loc.port || (loc.protocol === 'https:' ? 443 : 80); - const pathname = path || '/wsapi/in/'; - const fullUrl = url || `${host}:${portNumber}${pathname}`; + const host = overrides.hostname || loc.hostname; + const port = overrides.port || loc.port || (loc.protocol === 'https:' ? 443 : 80); + const target = overrides.url || `${host}:${port}${overrides.path || '/wsapi/in/'}`; + return `${protocol}${target}`; +} + +function backoffDelay (failures: number): number { + return Math.min(1_000 * 2 ** failures, 15 * 60_000); +} + +function parseFrame (data: string): Record { + const frame = JSON.parse(data) as unknown; + if (!frame || typeof frame !== 'object' || Array.isArray(frame)) { + throw new Error('WebSocket logger accepts JSON object frames only.'); + } + return frame as Record; +} - return `${protocol}${fullUrl}`; +function eventIdOf (data: unknown): string | null { + try { + const frame = typeof data === 'string' ? parseFrame(data) : data as Record; + const metadata = frame.metadata as Record | undefined; + return typeof metadata?.eventId === 'string' && metadata.eventId ? metadata.eventId : null; + } catch { + return null; + } } +/** Clone before stamping: a resend must retain its identity, while every newly + * constructed transport frame must receive a fresh one. */ +function stamped (frame: Record): string { + const copy = { + ...frame, + metadata: { ...((frame.metadata as Record | undefined) ?? {}) } + }; + util.timestampEvent(copy); + return JSON.stringify(copy); +} + +function normalizedEvent (data: string): string { + const frame = parseFrame(data); + return eventIdOf(frame) === null ? stamped(frame) : data; +} + +/** A sending gate. It controls leasing only; admission never waits here. */ +class Gate { + private open = false; + private waiters: Array<() => void> = []; + + set (open: boolean): void { + this.open = open; + if (open) this.waiters.splice(0).forEach(resolve => resolve()); + } -export function websocketLogger (server: string | WsHostOverrides = {}): Logger { - /* - This is a pretty complex logger, which sends events over a web - socket. + wait (): Promise { + return this.open + ? Promise.resolve() + : new Promise(resolve => { this.waiters.push(resolve); }); + } +} - `server` can be a URL (usually, ws:// or wss://) or an object - containing one or more of hostname, port, path, and url. +export function websocketLogger ( + server: string | WsHostOverrides = {}, + { + autoack = false, + namespace, + fetchState = !autoack, + queueType = QueueType.AUTODETECT as string + }: WebsocketLoggerOptions = {} +): Logger { + let serverUrl = typeof server === 'string' ? server : wsHost(server); + // Storage identity must not depend on whether a mutable server override is + // read before or after the first enqueue. Explicit/app namespaces supersede + // this stable construction-time fallback. + const defaultQueueNamespace = serverUrl; + let queueNamespace = namespace ?? null; + let queue: Queue | null = null; + const profile = autoack ? 'autoack' : 'durable'; + const outbox = (): Queue => { + const resolved = queueNamespace ?? defaultQueueNamespace; + return queue ??= new Queue(`lo-event:${encodeURIComponent(resolved)}:${profile}`, { queueType }); + }; - Note that if the server is an object, it can be overwritten in - storage (key loServer). + const engine = new DeliveryEngine({ autoack }); + const gate = new Gate(); + const lockedFields: Record = {}; - Most of the complexity comes from reconnections, retries, - etc. and the need to keep robust queues, as well as the need be - robust about queuing events before we have a socket open or during - a network failure. - */ + let SocketLibrary: SocketConstructor; let socket: WebSocket | null = null; - // Minimal WebSocket constructor — works with both browser WebSocket and the `ws` package - let WSLibrary: new (url: string) => WebSocket; - const queue = new Queue('websocketLogger'); - // This holds an exception, if we're blacklisted, between the web - // socket and the API. We generate this when we receive a message, - // which is not a helpful place to raise the exception from, so we - // keep this around until we're called from the client, and then we - // raise it there. - let blockerror: disabler.BlockError | null = null; - let metadata: Record = {}; - - // Resolve server to a URL string - let serverUrl: string; - if(!server) { - serverUrl = wsHost(); - } else if(typeof server === 'object') { - serverUrl = wsHost(server); - } else { - serverUrl = server; + let socketAttempt = 0; + let initialized = false; + let initialization: Promise | null = null; + let ticker: ReturnType | null = null; + let waitingOnDisabler = false; + let failCurrentConnection: (() => void) | null = null; + + /** External facts are reduced in arrival order. I/O answers start a new fact + * instead of holding the serial lane, so a hung maxSeq cannot block elapsed() + * from opening the barrier deadline. */ + let protocolWork: Promise = Promise.resolve(); + + function submit (fact: () => Decision[]): Promise { + const work = protocolWork.then(() => { perform(fact()); }); + protocolWork = work.catch(error => { + debug.error('websocketLogger: protocol executor failed', error); + }); + return work; } - function calculateExponentialBackoff (n: number) { - return Math.min(1000 * Math.pow(2, n), 1000 * 60 * 15); + function perform (decisions: Decision[]): void { + for (const decision of decisions) { + switch (decision.do) { + case 'rewind': + outbox().rewind(); + break; + + case 'measureWatermark': { + const generation = engine.generation(); + void outbox().maxSeq().then( + max => submit(() => engine.watermarkResult(generation, max)), + error => { + debug.error('websocketLogger: could not read outbox watermark', error); + return submit(() => engine.measurementFailed(generation, 'watermark measurement')); + } + ); + break; + } + + case 'probeQueue': { + const generation = engine.generation(); + void outbox().unleasedAtOrBelow(decision.watermark).then( + count => submit(() => engine.probeResult(generation, count)), + error => { + debug.error('websocketLogger: could not probe outbox', error); + return submit(() => engine.measurementFailed(generation, 'probe')); + } + ); + break; + } + + case 'sendFrame': { + const generation = engine.generation(); + if (sendNow(decision.frame)) { + perform(engine.sendCompleted(generation)); + } else { + perform(engine.sendFailed(generation)); + failCurrentConnection?.(); + } + break; + } + + case 'confirmIds': + outbox().confirm(decision.ids); + break; + + case 'askForState': { + const generation = engine.generation(); + if (sendNow(decision.frame)) { + perform(engine.stateSendCompleted(generation)); + } else { + perform(engine.stateSendFailed(generation)); + failCurrentConnection?.(); + } + break; + } + + case 'pauseSending': + gate.set(false); + break; + case 'resumeSending': + gate.set(true); + break; + case 'discardOutbox': + outbox().clear(); + break; + case 'log': + if (decision.level === 'error') debug.error(`websocketLogger: ${decision.message}`); + else debug.info(`websocketLogger: ${decision.message}`); + break; + } + } } - let failures = 0; - let READY = false; - let wsFailureResolve: (() => void) | null = null; - let wsFailurePromise: Promise | null = null; - let wsConnectedResolve: ((value: boolean) => void) | null = null; + function sendNow (frame: string): boolean { + if (!socket || socket.readyState !== SOCKET_OPEN) return false; + try { + socket.send(frame); + return true; + } catch (error) { + debug.error('websocketLogger: socket send failed', error); + return false; + } + } - async function startWebsocketConnectionLoop () { + async function leaseLoop (): Promise { while (true) { - const connected = await newWebsocket(); - if (!connected) { - failures++; - await util.delay(calculateExponentialBackoff(failures)); - } else { - READY = true; - failures = 0; - await socketClosed(); - READY = false; + try { + await gate.wait(); + const leaseGeneration = engine.generation(); + const { seq, item } = await outbox().leaseNext(); + const frame = typeof item === 'string' ? item : JSON.stringify(item); + let rewound = false; + await submit(() => { + // A lease may have been parked since the previous connection. Its + // cursor observation is stale even if a new socket is now online. + const decisions: Decision[] = leaseGeneration === engine.generation() + ? engine.recordLeased(seq, eventIdOf(item), frame) + : [{ do: 'rewind' }]; + rewound = decisions.some(decision => decision.do === 'rewind'); + return decisions; + }); + if (rewound) { + await util.delay(RETRY_PAUSE_MS); + } + } catch (error) { + // A transient storage failure must not kill the only delivery loop. + debug.error('websocketLogger: outbox lease loop failed; retrying', error); + await util.delay(RETRY_PAUSE_MS); } } } - function socketClosed () { return wsFailurePromise; } + function enqueueMetadataPreamble (): void { + if (!disabler.storeEvents() || !Object.keys(lockedFields).length) return; + outbox().enqueue(stamped({ event: 'lock_fields', fields: { ...lockedFields } })); + } - function newWebsocket () { - socket = new WSLibrary(serverUrl); - wsFailurePromise = new Promise((resolve) => { - wsFailureResolve = resolve; - }); - const wsConnectedPromise = new Promise((resolve) => { - wsConnectedResolve = resolve; - }); - socket.onopen = () => { prepareSocket(); wsConnectedResolve!(true); }; - socket.onerror = function (e) { - debug.error('Could not connect to websocket', e); - wsConnectedResolve!(false); - wsFailureResolve!(); - }; - socket.onclose = () => { wsConnectedResolve!(false); wsFailureResolve!(); }; - socket.onmessage = receiveMessage; - return wsConnectedPromise; + function startTicker (attempt: number): void { + stopTicker(); + let lastTick = Date.now(); + ticker = setInterval(() => { + if (attempt !== socketAttempt) return; + const now = Date.now(); + const elapsed = now - lastTick; + lastTick = now; + void submit(() => engine.elapsed(elapsed)); + }, TICK_MS); + (ticker as ReturnType & { unref?: () => void }).unref?.(); } - function prepareSocket () { - if(Object.keys(metadata).length > 0) { - queue.enqueue(JSON.stringify(metadata)); - } + function stopTicker (): void { + if (ticker !== null) clearInterval(ticker); + ticker = null; } - async function socketSend (item: unknown) { - socket!.send(item as string); + function runConnection (): Promise { + return new Promise(resolve => { + const attempt = ++socketAttempt; + const candidate = new SocketLibrary(serverUrl); + socket = candidate; + let opened = false; + let settled = false; + let invalidated = false; + + const finish = () => { + if (settled) return; + settled = true; + if (attempt === socketAttempt) { + stopTicker(); + if (!invalidated) { + invalidated = true; + void submit(() => engine.disconnected()); + util.dispatchCustomEvent('lo_connection_status', { detail: { connected: false } }); + } + if (socket === candidate) socket = null; + if (failCurrentConnection === fail) failCurrentConnection = null; + } + resolve(opened); + }; + + const fail = () => { + if (attempt !== socketAttempt || invalidated) return; + invalidated = true; + stopTicker(); + // Invalidate this connection synchronously. Waiting for the browser's + // close event would leave a window in which an already-computed queue + // probe could still be accepted under this connection's generation. + perform(engine.disconnected()); + util.dispatchCustomEvent('lo_connection_status', { detail: { connected: false } }); + try { candidate.close(); } catch { finish(); } + }; + + failCurrentConnection = fail; + + candidate.onopen = () => { + if (attempt !== socketAttempt) return; + opened = true; + try { + enqueueMetadataPreamble(); + void submit(() => engine.connected()).then(() => { + if (attempt !== socketAttempt || candidate.readyState !== SOCKET_OPEN) return; + startTicker(attempt); + util.dispatchCustomEvent('lo_connection_status', { detail: { connected: true } }); + }).catch(error => { + debug.error('websocketLogger: failed to prepare an open connection', error); + try { candidate.close(); } catch { /* best effort */ } + }); + } catch (error) { + debug.error('websocketLogger: failed to prepare an open connection', error); + try { candidate.close(); } catch { /* best effort */ } + finish(); + } + }; + candidate.onmessage = event => receiveMessage(event, attempt); + candidate.onclose = finish; + candidate.onerror = event => { + debug.error('websocketLogger: websocket error', event); + try { candidate.close(); } catch { /* already closed */ } + finish(); + }; + }); } - async function waitForWSReady () { - return await util.backoff( - () => (READY), - 'WebSocket not ready', - undefined, - util.TERMINATION_POLICY.RETRY - ); + async function connectionLoop (): Promise { + let failures = 0; + while (true) { + try { + const opened = await runConnection(); + failures = opened ? 0 : failures + 1; + } catch (error) { + failures++; + debug.error('websocketLogger: connection loop failed', error); + } + await util.delay(backoffDelay(failures)); + } } - function receiveMessage (event: MessageEvent) { - const response = JSON.parse(event.data); + function receiveMessage (event: MessageEvent, attempt: number): void { + let response: Record; + try { + response = JSON.parse(String(event.data)); + } catch (error) { + debug.error('websocketLogger: ignoring invalid JSON from server', error); + return; + } + + // These are world facts: an old socket's durable ack or snapshot response + // is still useful. Other side-channel frames belong to their connection. + if (response.status === 'ack') { + if (typeof response.id === 'string') void submit(() => engine.ackReceived(response.id)); + return; + } + if (response.status === 'fetch_blob') { + void submit(() => engine.stateReceived()); + util.dispatchCustomEvent('fetch_blob', { detail: response.data }); + return; + } + if (attempt !== socketAttempt) return; + switch (response.status) { - case 'blocklist': - debug.info('Received block error from server'); - blockerror = new disabler.BlockError( - response.message, - response.time_limit, - response.action - ); + case 'blocklist': { + const block = new disabler.BlockError(response.message, response.time_limit, response.action); + disabler.handleBlockError(block); + if (!disabler.streamEvents()) { + void submit(() => engine.disablerEngaged({ + permanent: disabler.isPermanent(), + permanentOptOut: disabler.isPermanentOptOut() + })); + if (!disabler.isPermanent()) void awaitDisablerRelease(); + } break; + } case 'auth': { - // Server pushes identity after it resolves the WS auth (HTTP Basic via - // nginx, LTI session, guest cookie, etc.). We stash it in the storage - // shim (for non-Redux consumers) and dispatch a DOM CustomEvent so - // reduxLogger (and anything else that listens) can react. - // - // Forward-compat: we spread every field except `status` into the user - // object so new profile fields (avatar, role, safe_user_id, ...) added - // server-side flow through without touching this file. Consumers - // should treat `user_id` as the only required field. - const { status, ...user } = response; + const { status: _status, ...user } = response; storage.set(user); util.dispatchCustomEvent('auth', { detail: user }); break; } - // These should probably be behind a feature flag, as they assume - // we trust the server. case 'local_storage': storage.set({ [response.key]: response.value }); break; case 'browser_event': util.dispatchCustomEvent(response.event_type, { detail: response.detail }); break; - case 'fetch_blob': - util.dispatchCustomEvent('fetch_blob', { detail: response.data }); + case 'save_blob_ack': + util.dispatchCustomEvent('save_blob_ack', { detail: { token: response.token } }); break; - default: - debug.info(`Received response we do not yet handle: ${JSON.stringify(response)}`); + case 'save_blob_nack': + util.dispatchCustomEvent('save_blob_nack', { detail: { token: response.token } }); break; + default: + debug.info(`websocketLogger: unhandled server frame: ${JSON.stringify(response)}`); } } - function checkForBlockError () { - if (blockerror) { - console.log('Throwing block error'); - const b = blockerror; - blockerror = null; - socket!.close(); - throw b; + async function awaitDisablerRelease (): Promise { + if (waitingOnDisabler) return; + waitingOnDisabler = true; + try { + if (await disabler.retry()) await submit(() => engine.disablerReleased()); + } catch (error) { + debug.error('websocketLogger: disabler wait failed; delivery remains paused', error); + } finally { + waitingOnDisabler = false; } } - function wsLogData (data: string) { - checkForBlockError(); - queue.enqueue(data); - } + const logger = ((data: string) => { + if (!disabler.storeEvents()) return; + const frame = parseFrame(data); + if (util.isProtocolEventName(String(frame.event))) { + throw new Error(`Application event name is reserved: ${String(frame.event)}`); + } + outbox().enqueue(normalizedEvent(data)); + }) as Logger; + + logger.configure = ({ source }) => { + if (initialized) throw new Error('WebSocket logger cannot be configured after init.'); + if (namespace === undefined) { + if (queue !== null) throw new Error('WebSocket logger was used before its application namespace was configured.'); + queueNamespace = source; + } + }; - wsLogData.init = async function () { - // Check storage for server override (the storage API is callback-based, - // so this must happen in async context, not at construction time) + logger.init = () => initialization ??= (async () => { + initialized = true; try { - const stored = await new Promise(resolve => storage.get('lo_server', resolve)); - if (stored && (stored as Record).lo_server) { - debug.info('Overriding server from storage'); - serverUrl = (stored as Record).lo_server as string; - } - } catch (e) { - debug.info('Could not check storage for server override'); + const stored = await new Promise>(resolve => storage.get('lo_server', resolve)); + if (typeof stored.lo_server === 'string') serverUrl = stored.lo_server; + } catch (error) { + debug.info(`websocketLogger: could not read server override: ${String(error)}`); } - if (typeof WebSocket === 'undefined') { - debug.info('Importing ws'); - WSLibrary = (await import('ws')).WebSocket as unknown as new (url: string) => WebSocket; - } else { - debug.info('Using built-in websocket'); - WSLibrary = WebSocket; + SocketLibrary = typeof WebSocket === 'undefined' + ? (await import('ws')).WebSocket as unknown as SocketConstructor + : WebSocket; + + if (!disabler.streamEvents()) { + await submit(() => engine.disablerEngaged({ + permanent: disabler.isPermanent(), + permanentOptOut: disabler.isPermanentOptOut() + })); + if (!disabler.isPermanent()) void awaitDisablerRelease(); } - startWebsocketConnectionLoop(); - queue.startDequeueLoop({ - initialize: waitForWSReady, - shouldDequeue: waitForWSReady, - onDequeue: socketSend - }); - }; + if (fetchState) await submit(() => engine.requestState(FETCH_BLOB_FRAME)); + void connectionLoop().catch(error => debug.error('websocketLogger: connection loop stopped', error)); + void leaseLoop().catch(error => debug.error('websocketLogger: lease loop stopped', error)); + })(); - wsLogData.setField = function (data: string) { - util.mergeDictionary(metadata, JSON.parse(data)); - queue.enqueue(data); + logger.setField = data => { + if (!disabler.storeEvents()) return; + const frame = parseFrame(data); + const fields = frame.fields; + if (fields && typeof fields === 'object' && !Array.isArray(fields)) { + util.mergeDictionary(lockedFields, fields as Record); + } + outbox().enqueue(normalizedEvent(data)); }; - function handleSaveBlob (blob: unknown) { - queue.enqueue(JSON.stringify({ event: 'save_blob', blob })); - } + logger.requestState = () => { void submit(() => engine.requestState(FETCH_BLOB_FRAME)); }; + logger.unackedCount = () => outbox().unconfirmedCount(); + logger.queueDebug = { + count: () => outbox().unconfirmedCount(), + inspect: (limit = 20) => outbox().inspect(limit), + clear: () => outbox().clear() + }; + logger.lo_name = 'Reliable WebSocket Logger'; + logger.lo_id = 'websocket_logger'; - util.consumeCustomEvent('save_blob', handleSaveBlob); + util.consumeCustomEvent('save_blob', (data: unknown) => { + if (!disabler.storeEvents()) return; + const { blob, token } = data as { blob: unknown; token: number }; + outbox().enqueue(stamped({ event: 'save_blob', blob, token })); + }); - return wsLogData as Logger; + return logger; } diff --git a/tests/indexeddbQueue.test.js b/tests/indexeddbQueue.test.js new file mode 100644 index 0000000..f47d3aa --- /dev/null +++ b/tests/indexeddbQueue.test.js @@ -0,0 +1,97 @@ +import { beforeAll, describe, expect, it } from 'vitest'; +import { IDBKeyRange, indexedDB } from 'fake-indexeddb'; +import { Queue } from '../src/indexeddbQueue.js'; +import { queueContract } from './queueContract.js'; + +beforeAll(() => { + globalThis.indexedDB = indexedDB; + globalThis.IDBKeyRange = IDBKeyRange; +}); + +const name = label => `reliable-delivery-${label}-${crypto.randomUUID()}`; + +queueContract('IndexedDB', label => new Queue(name(`shared-${label}`))); + +describe('IndexedDB outbox contract', () => { + it('leases without deleting and confirms only explicit ids', async () => { + const queue = new Queue(name('lease')); + queue.enqueue('a'); + queue.enqueue('b'); + expect(await queue.leaseNext()).toEqual({ seq: 1, item: 'a' }); + expect(await queue.leaseNext()).toEqual({ seq: 2, item: 'b' }); + expect(await queue.unconfirmedCount()).toBe(2); + queue.confirm([1]); + expect(await queue.unconfirmedCount()).toBe(1); + queue.rewind(); + expect(await queue.leaseNext()).toEqual({ seq: 2, item: 'b' }); + }); + + it('a parked lease wakes by rescanning committed storage', async () => { + const queue = new Queue(name('park')); + const waiting = queue.leaseNext(); + queue.enqueue('later'); + expect(await waiting).toEqual({ seq: 1, item: 'later' }); + }); + + it('polling discovers an enqueue committed by another context', async () => { + const database = name('cross-tab'); + const firstTab = new Queue(database); + const secondTab = new Queue(database); + const waiting = firstTab.leaseNext(); + secondTab.enqueue('other-tab'); + await secondTab.unconfirmedCount(); + expect(await waiting).toEqual({ seq: 1, item: 'other-tab' }); + }); + + it('a parked lease wakes with the lowest stored record, not the local waker', async () => { + const database = name('lowest-on-wake'); + const firstTab = new Queue(database); + const secondTab = new Queue(database); + firstTab.enqueue('first'); + await firstTab.unconfirmedCount(); + await firstTab.leaseNext(); + + const waiting = firstTab.leaseNext(); + secondTab.enqueue('lower-cross-tab-record'); + await secondTab.unconfirmedCount(); + firstTab.enqueue('local-waker'); + + expect(await waiting).toEqual({ seq: 2, item: 'lower-cross-tab-record' }); + }); + + it('a rewind racing an in-progress scan cannot be overwritten', async () => { + const queue = new Queue(name('rewind-race')); + queue.enqueue('first'); + queue.enqueue('second'); + expect((await queue.leaseNext()).seq).toBe(1); + + const racingLease = queue.leaseNext(); + queue.rewind(); + expect(await racingLease).toEqual({ seq: 1, item: 'first' }); + }); + + it('a rewind racing a barrier count cannot make the backlog look leased', async () => { + const queue = new Queue(name('barrier-rewind-race')); + queue.enqueue('first'); + queue.enqueue('second'); + await queue.leaseNext(); + await queue.leaseNext(); + + const racingCount = queue.unleasedAtOrBelow(2); + queue.rewind(); + expect(await racingCount).toBe(2); + }); + + it('one sender cannot range-delete another sender\'s records', async () => { + const database = name('explicit-confirm'); + const firstTab = new Queue(database); + const secondTab = new Queue(database); + for (const item of ['A1', 'B1', 'A2', 'B2']) firstTab.enqueue(item); + await firstTab.unconfirmedCount(); + secondTab.confirm([1, 3]); + expect(await secondTab.unconfirmedCount()).toBe(2); + secondTab.rewind(); + expect(await secondTab.leaseNext()).toEqual({ seq: 2, item: 'B1' }); + expect(await secondTab.leaseNext()).toEqual({ seq: 4, item: 'B2' }); + }); +}); diff --git a/tests/loEventFanout.test.js b/tests/loEventFanout.test.js new file mode 100644 index 0000000..d49e614 --- /dev/null +++ b/tests/loEventFanout.test.js @@ -0,0 +1,25 @@ +import { describe, expect, it, vi } from 'vitest'; +import * as loEvent from '../src/loEvent.js'; + +describe('loEvent front desk', () => { + it('configures application identity and isolates throwing sibling loggers', async () => { + const contexts = []; + const received = []; + const broken = Object.assign(() => { throw new Error('expected logger failure'); }, { + lo_id: 'broken' + }); + const healthy = Object.assign(event => { received.push(JSON.parse(event)); }, { + configure: context => { contexts.push(context); }, + lo_id: 'healthy' + }); + + loEvent.init('fanout-test', '1', [broken, healthy], { useDisabler: false }); + loEvent.go(); + loEvent.logEvent('answer', { value: 42 }); + + await vi.waitFor(() => { + expect(received.some(event => event.event === 'answer')).toBe(true); + }); + expect(contexts).toEqual([{ source: 'fanout-test' }]); + }); +}); diff --git a/tests/lo_event.test.js b/tests/lo_event.test.js index 8c76a21..6ae2daf 100644 --- a/tests/lo_event.test.js +++ b/tests/lo_event.test.js @@ -8,6 +8,7 @@ import * as reduxLogger from '../src/reduxLogger.js'; import { consoleLogger } from '../src/consoleLogger.js'; import * as debug from '../src/debugLog.js'; import { getBrowserInfo } from '../src/metadata/browserinfo.js'; +import * as disabler from '../src/disabler.js'; const rl = reduxLogger.reduxLogger(); @@ -45,4 +46,24 @@ describe('loEvent testing', () => { expect(fields.version).toBe('1'); expect(fields.preauth_type).toBe('test'); }); + + it('rejects application use of protocol-reserved frame names', () => { + expect(() => loEvent.logEvent('lock_fields', {})).toThrow(/reserved/); + expect(() => loEvent.logEvent('fetch_blob', {})).toThrow(/reserved/); + }); + + it('moves events through the front desk while transmission is blocked', async () => { + disabler.handleBlockError(new disabler.BlockError( + 'retain locally', + 'PERMANENT', + 'MAINTAIN' + )); + loEvent.logEvent('blocked-admission', { marker: 99 }); + + let received; + do { + received = await reduxLogger.awaitEvent(); + } while (received.event !== 'blocked-admission'); + expect(received.marker).toBe(99); + }); }); diff --git a/tests/protocol.test.js b/tests/protocol.test.js new file mode 100644 index 0000000..7b11108 --- /dev/null +++ b/tests/protocol.test.js @@ -0,0 +1,262 @@ +import { describe, expect, it } from 'vitest'; +import { + BARRIER_DEADLINE_MS, + DeliveryEngine, + PROBE_INTERVAL_MS, + SNAPSHOT_RETRY_MS +} from '../src/protocol.js'; + +const FETCH = JSON.stringify({ event: 'fetch_blob' }); +const kinds = decisions => decisions.map(decision => decision.do); +const pick = (decisions, kind) => decisions.filter(decision => decision.do === kind); + +function connectedEngine (options = {}) { + const engine = new DeliveryEngine(options); + engine.connected(); + engine.watermarkResult(engine.generation(), null); + return engine; +} + +function send (engine, seq, id) { + const leased = engine.recordLeased(seq, id, JSON.stringify({ metadata: { eventId: id } })); + return [...leased, ...engine.sendCompleted(engine.generation())]; +} + +describe('connection and confirmation', () => { + it('rewinds before enabling sends or measuring the watermark', () => { + expect(kinds(new DeliveryEngine().connected())) + .toEqual(['rewind', 'resumeSending', 'measureWatermark']); + }); + + it('invalidates every asynchronous connection result on close', () => { + const engine = new DeliveryEngine(); + engine.connected(); + const stale = engine.generation(); + engine.disconnected(); + engine.connected(); + expect(engine.watermarkResult(stale, null)).toEqual([]); + expect(engine.probeResult(stale, 0)).toEqual([]); + expect(engine.sendCompleted(stale)).toEqual([]); + }); + + it('durable mode confirms exactly the identity the server acked', () => { + const engine = connectedEngine(); + expect(pick(send(engine, 7, 'browser.session.7'), 'confirmIds')).toEqual([]); + expect(engine.ackReceived('browser.session.7')) + .toEqual([{ do: 'confirmIds', ids: [7] }]); + expect(engine.ackReceived('browser.session.7')).toEqual([]); + }); + + it('registers identity before send and keeps it across reconnects', () => { + const engine = connectedEngine(); + engine.recordLeased(7, 'browser.session.7', '{}'); + engine.disconnected(); + engine.connected(); + expect(engine.ackReceived('browser.session.7')) + .toContainEqual({ do: 'confirmIds', ids: [7] }); + }); + + it('autoack confirms only after a successful send', () => { + const engine = connectedEngine({ autoack: true }); + engine.recordLeased(7, 'browser.session.7', '{}'); + expect(engine.sendFailed(engine.generation()).some(d => d.do === 'confirmIds')).toBe(false); + + engine.recordLeased(7, 'browser.session.7', '{}'); + expect(engine.sendCompleted(engine.generation())) + .toContainEqual({ do: 'confirmIds', ids: [7] }); + }); + + it('a failed send removes its unsent identity from acknowledgement tracking', () => { + const engine = connectedEngine(); + engine.recordLeased(7, 'browser.session.7', '{}'); + expect(engine.awaitingAck()).toBe(1); + + engine.sendFailed(engine.generation()); + expect(engine.awaitingAck()).toBe(0); + expect(engine.ackReceived('browser.session.7')).toEqual([]); + }); + + it('drains an unnamed legacy record loudly after send', () => { + const decisions = send(connectedEngine(), 7, null); + expect(pick(decisions, 'log')[0].level).toBe('error'); + expect(decisions).toContainEqual({ do: 'confirmIds', ids: [7] }); + }); +}); + +describe('disabler policy', () => { + it('a block pauses sends and a parked lease is rewound', () => { + const engine = connectedEngine(); + expect(kinds(engine.disablerEngaged())).toEqual(['pauseSending']); + expect(kinds(engine.recordLeased(7, 'id', '{}'))).toEqual(['rewind']); + expect(kinds(engine.disablerReleased())).toEqual(['resumeSending']); + }); + + it('permanent MAINTAIN retains the outbox, while privacy DROP discards it', () => { + const maintain = connectedEngine().disablerEngaged({ permanent: true }); + expect(kinds(maintain)).not.toContain('discardOutbox'); + expect(pick(maintain, 'log')[0].level).toBe('error'); + + const drop = connectedEngine().disablerEngaged({ permanent: true, permanentOptOut: true }); + expect(kinds(drop)).toContain('discardOutbox'); + }); +}); + +describe('flush barrier', () => { + it('un-evaluated refuses the snapshot until measurement clears it', () => { + const engine = new DeliveryEngine(); + engine.connected(); + expect(engine.requestState(FETCH)).toEqual([]); + expect(engine.watermarkResult(engine.generation(), null)) + .toEqual([{ do: 'askForState', frame: FETCH }]); + }); + + it('does not clear from a probe in the lease-to-send window', () => { + const engine = new DeliveryEngine(); + engine.connected(); + const generation = engine.generation(); + engine.watermarkResult(generation, 7); + engine.recordLeased(7, 'id', '{}'); + expect(engine.probeResult(generation, 0)).toEqual([]); + expect(engine.barrierIsClear()).toBe(false); + expect(engine.sendCompleted(generation)) + .toContainEqual({ do: 'probeQueue', watermark: 7 }); + }); + + it('fallback probes notice deletions performed by another tab', () => { + const engine = new DeliveryEngine(); + engine.connected(); + engine.watermarkResult(engine.generation(), 7); + expect(engine.elapsed(PROBE_INTERVAL_MS)) + .toContainEqual({ do: 'probeQueue', watermark: 7 }); + }); + + it('deadline covers a maxSeq measurement that never settles', () => { + const engine = new DeliveryEngine(); + engine.connected(); + engine.requestState(FETCH); + const decisions = engine.elapsed(BARRIER_DEADLINE_MS); + expect(pick(decisions, 'log')[0].level).toBe('error'); + expect(decisions).toContainEqual({ do: 'askForState', frame: FETCH }); + }); + + it('a shrinking backlog holds the barrier open past the deadline', () => { + const engine = new DeliveryEngine(); + engine.connected(); + engine.requestState(FETCH); + expect(pick(engine.elapsed(BARRIER_DEADLINE_MS - 1), 'askForState')).toEqual([]); + const generation = engine.generation(); + engine.watermarkResult(generation, 10); + engine.probeResult(generation, 10); + + expect(pick(engine.elapsed(BARRIER_DEADLINE_MS - 1), 'askForState')).toEqual([]); + engine.probeResult(generation, 9); + expect(pick(engine.elapsed(BARRIER_DEADLINE_MS - 1), 'askForState')).toEqual([]); + expect(engine.barrierIsClear()).toBe(false); + }); + + it('a backlog that stops shrinking still opens at the deadline', () => { + const engine = new DeliveryEngine(); + engine.connected(); + engine.requestState(FETCH); + const generation = engine.generation(); + engine.watermarkResult(generation, 10); + engine.probeResult(generation, 10); + + expect(engine.elapsed(BARRIER_DEADLINE_MS)) + .toContainEqual({ do: 'askForState', frame: FETCH }); + }); + + it('measurement failure opens loudly rather than hanging', () => { + const engine = new DeliveryEngine(); + engine.connected(); + engine.requestState(FETCH); + const decisions = engine.measurementFailed(engine.generation(), 'watermark'); + expect(pick(decisions, 'log')[0].level).toBe('error'); + expect(kinds(decisions)).toContain('askForState'); + }); +}); + +describe('state snapshot', () => { + it('starts the retry clock only after the direct send succeeds', () => { + const engine = connectedEngine(); + expect(kinds(engine.requestState(FETCH))).toEqual(['askForState']); + expect(engine.elapsed(SNAPSHOT_RETRY_MS * 2)).toEqual([]); + + engine.stateSendCompleted(engine.generation()); + expect(engine.elapsed(SNAPSHOT_RETRY_MS - 1)).toEqual([]); + expect(kinds(engine.elapsed(1))).toEqual(['log', 'askForState']); + }); + + it('re-asks on reconnect, stops when fulfilled, and allows a new request', () => { + const engine = connectedEngine(); + engine.requestState(FETCH); + engine.stateSendCompleted(engine.generation()); + engine.disconnected(); + engine.connected(); + expect(engine.watermarkResult(engine.generation(), null)) + .toContainEqual({ do: 'askForState', frame: FETCH }); + + engine.stateReceived(); + engine.disconnected(); + engine.connected(); + expect(engine.watermarkResult(engine.generation(), null)).toEqual([]); + expect(engine.requestState(FETCH)).toContainEqual({ do: 'askForState', frame: FETCH }); + }); + + it('accepts a response as a world fact even after its socket changed', () => { + const engine = connectedEngine(); + engine.requestState(FETCH); + engine.disconnected(); + engine.stateReceived(); + engine.connected(); + expect(engine.watermarkResult(engine.generation(), null)).toEqual([]); + }); +}); + +describe('annotated wire trace', () => { + it('sends the crash backlog before the snapshot and resends unacked work', () => { + const engine = new DeliveryEngine(); + const sent = []; + const confirmed = []; + const track = decisions => { + sent.push(...pick(decisions, 'sendFrame').map(decision => decision.seq)); + sent.push(...pick(decisions, 'askForState').map(() => 'fetch_blob')); + confirmed.push(...pick(decisions, 'confirmIds').flatMap(decision => decision.ids)); + return decisions; + }; + + track(engine.connected()); + track(engine.requestState(FETCH)); + const first = engine.generation(); + track(engine.watermarkResult(first, 42)); + + track(engine.recordLeased(41, 'B.S1.38', '{"event":"save_blob"}')); + track(engine.sendCompleted(first)); + track(engine.probeResult(first, 1)); + track(engine.recordLeased(42, 'B.S1.39', '{"event":"answer"}')); + track(engine.sendCompleted(first)); + track(engine.probeResult(first, 0)); + track(engine.stateSendCompleted(first)); + + track(engine.recordLeased(43, 'B.S2.1', '{"event":"keystroke"}')); + track(engine.sendCompleted(first)); + track(engine.ackReceived('B.S1.38')); + track(engine.ackReceived('B.S1.39')); + track(engine.stateReceived()); + + expect(sent).toEqual([41, 42, 'fetch_blob', 43]); + expect(confirmed).toEqual([41, 42]); + + track(engine.disconnected()); + track(engine.connected()); + const second = engine.generation(); + track(engine.watermarkResult(second, 43)); + track(engine.recordLeased(43, 'B.S2.1', '{"event":"keystroke"}')); + track(engine.sendCompleted(second)); + track(engine.probeResult(second, 0)); + track(engine.ackReceived('B.S2.1')); + + expect(sent).toEqual([41, 42, 'fetch_blob', 43, 43]); + expect(confirmed).toEqual([41, 42, 43]); + }); +}); diff --git a/tests/queue.test.js b/tests/queue.test.js index 8057af8..629074f 100644 --- a/tests/queue.test.js +++ b/tests/queue.test.js @@ -1,8 +1,9 @@ -// TODO: Test both types of queue, and then in node and -// browser, as well as various failure conditions. - import { describe, it, expect } from 'vitest'; import { Queue } from '../src/queue.js'; +import { Queue as MemoryQueue } from '../src/memoryQueue.js'; +import { queueContract } from './queueContract.js'; + +queueContract('memory', label => new MemoryQueue(`shared-${label}-${crypto.randomUUID()}`)); describe('Queue', () => { it('dequeues items in FIFO order', async () => { @@ -40,3 +41,197 @@ describe('Queue', () => { expect(received).toEqual(['a', 'b']); }); }); + +// TODO(final PR review): these lease/confirm/rewind cases were load-bearing +// while building the lease discipline (caught the rewind-parked-consumer bug), +// but the algorithm is now stable — decide whether to keep all of them, trim to +// the two that pin the contract (lease-doesn't-delete + cumulative-confirm), or +// pull. They are mock-free/declarative, so low weight, but per the testing +// philosophy a stable algorithm's tests are candidate maintenance weight. +// +// Ack-protocol backbone: lease is non-destructive; confirm deletes the +// acked prefix; rewind re-hands unconfirmed items (resend on reconnect). +describe('MemoryQueue lease / confirm / rewind', () => { + it('leases with seq WITHOUT deleting; confirm deletes exactly what it is given', async () => { + const q = new MemoryQueue('lease-confirm'); + q.enqueue('a'); q.enqueue('b'); q.enqueue('c'); + + expect(await q.leaseNext()).toEqual({ seq: 1, item: 'a' }); + expect(await q.leaseNext()).toEqual({ seq: 2, item: 'b' }); + expect(q.unconfirmedCount()).toBe(3); // leased, but nothing deleted + + q.confirm([1, 2]); // the two records this sender sent + expect(q.unconfirmedCount()).toBe(1); // only 'c' remains + expect(await q.leaseNext()).toEqual({ seq: 3, item: 'c' }); + }); + + it('rewind re-hands every unconfirmed item (full resend)', async () => { + const q = new MemoryQueue('rewind-all'); + q.enqueue('a'); q.enqueue('b'); + await q.leaseNext(); await q.leaseNext(); // sent, not acked + + q.rewind(); // reconnect + + expect(await q.leaseNext()).toEqual({ seq: 1, item: 'a' }); + expect(await q.leaseNext()).toEqual({ seq: 2, item: 'b' }); + }); + + it('confirm then rewind: only the unconfirmed tail resends', async () => { + const q = new MemoryQueue('rewind-partial'); + q.enqueue('a'); q.enqueue('b'); q.enqueue('c'); + await q.leaseNext(); await q.leaseNext(); await q.leaseNext(); + + q.confirm([1]); // 'a' durably acked + q.rewind(); // reconnect + + expect(await q.leaseNext()).toEqual({ seq: 2, item: 'b' }); + expect(q.unconfirmedCount()).toBe(2); + }); + + it('leaseNext parks when fully leased; rewind wakes the parked consumer', async () => { + const q = new MemoryQueue('park-rewind'); + q.enqueue('a'); + expect(await q.leaseNext()).toEqual({ seq: 1, item: 'a' }); + + let resolved = null; + const pending = q.leaseNext().then(v => { resolved = v; }); + await new Promise(r => setTimeout(r, 10)); + expect(resolved).toBe(null); // parked — nothing new to lease + + q.rewind(); // reconnect re-hands 'a' + await pending; + expect(resolved).toEqual({ seq: 1, item: 'a' }); + }); + + it('leaseNext parks then resolves on a later enqueue', async () => { + const q = new MemoryQueue('park-enqueue'); + let resolved = null; + const pending = q.leaseNext().then(v => { resolved = v; }); + await new Promise(r => setTimeout(r, 10)); + expect(resolved).toBe(null); + + q.enqueue('late'); + await pending; + expect(resolved).toEqual({ seq: 1, item: 'late' }); + }); +}); + +// ───────────────────────────────────────────────────────────────────────────── +// The shared-store, multi-sender hazard +// ───────────────────────────────────────────────────────────────────────────── +// +// One IndexedDB queue is shared by every tab in the browser — that sharing is +// exactly what makes tab-close recovery work. But each tab acks over its OWN +// socket. When deletion was a cumulative range (`delete(id <= n)`), one tab's +// ack deleted records belonging to other tabs, including records nobody had +// sent yet. Duplicates are covered by at-least-once delivery; deletions are +// not — that is silent data loss. +// +// The rule these lock in: a sender may delete ONLY the records it sent and saw +// acked on its own connection. + +describe('shared store, independent senders', () => { + it('one sender\'s ack does not delete another sender\'s unsent records', async () => { + const q = new MemoryQueue('two-tabs'); + // Interleaved, as two tabs writing to one store would be. + q.enqueue('A1'); q.enqueue('B1'); q.enqueue('A2'); q.enqueue('B2'); + + // Tab A sent only its own two records (seqs 1 and 3) and got them acked. + q.confirm([1, 3]); + + // Tab B's records must still be there. Under the old cumulative delete, + // confirming "through 3" would have taken B1 with it — unsent and gone. + expect(q.unconfirmedCount()).toBe(2); + q.rewind(); + expect(await q.leaseNext()).toEqual({ seq: 2, item: 'B1' }); + expect(await q.leaseNext()).toEqual({ seq: 4, item: 'B2' }); + }); + + it('a sender that dies before its ack loses nothing', async () => { + const q = new MemoryQueue('dead-tab'); + q.enqueue('x'); q.enqueue('y'); + await q.leaseNext(); await q.leaseNext(); // sent, never acked + + // The tab dies: its sent-map dies with it, so nothing is confirmed. + q.rewind(); // next connection + + expect(q.unconfirmedCount()).toBe(2); + expect(await q.leaseNext()).toEqual({ seq: 1, item: 'x' }); + }); +}); + +// ───────────────────────────────────────────────────────────────────────────── +// The flush barrier (snapshot-after-flush) +// ───────────────────────────────────────────────────────────────────────────── +// +// Before requesting a state snapshot, a connection must know its backlog has +// reached the server. The barrier is a question about the SHARED store, so it +// is asked of the store: capture the highest stored seq at connection start +// (maxSeq — the watermark), then probe unleasedAtOrBelow(watermark) until it +// reaches zero. Counting sends instead was wrong twice: on a shared store, +// another tab can send-and-delete records this connection was measured +// against, so no captured count is a quota this connection can be relied on +// to meet. + +describe('flush barrier (maxSeq / unleasedAtOrBelow)', () => { + it('maxSeq is null on an empty store, and the watermark otherwise', async () => { + const q = new MemoryQueue('barrier-empty'); + expect(await q.maxSeq()).toBe(null); + q.enqueue('a'); q.enqueue('b'); + expect(await q.maxSeq()).toBe(2); + }); + + it('clears as this connection leases (sends) the backlog', async () => { + const q = new MemoryQueue('barrier-drain'); + q.enqueue('a'); q.enqueue('b'); q.enqueue('c'); + const w = await q.maxSeq(); + + expect(await q.unleasedAtOrBelow(w)).toBe(3); + await q.leaseNext(); + await q.leaseNext(); + expect(await q.unleasedAtOrBelow(w)).toBe(1); + await q.leaseNext(); + expect(await q.unleasedAtOrBelow(w)).toBe(0); // barrier clear + }); + + it('ignores records enqueued after the watermark (live typing cannot starve it)', async () => { + const q = new MemoryQueue('barrier-live'); + q.enqueue('backlog'); + const w = await q.maxSeq(); + q.enqueue('keystroke-1'); q.enqueue('keystroke-2'); + + await q.leaseNext(); // the one backlog record + expect(await q.unleasedAtOrBelow(w)).toBe(0); // clear despite new events + }); + + it("clears when ANOTHER tab drains records this connection never leases", async () => { + // Sol's starvation case, the one no send count can handle: the store is + // shared, so another tab can send a backlog record and delete it on ack + // before this connection reaches it. A connection waiting to observe N of + // its own sends waits forever; asking the store instead sees the records + // gone — and gone-by-ack means the server already has them, which is + // exactly what the barrier wants to know. + const q = new MemoryQueue('barrier-cross-tab'); + q.enqueue('a'); q.enqueue('b'); q.enqueue('c'); + const w = await q.maxSeq(); + + await q.leaseNext(); // this connection sends 'a'... + q.confirm([2, 3]); // ...another tab sent and acked 'b' and 'c' + + expect(await q.unleasedAtOrBelow(w)).toBe(0); // barrier clear, no starvation + }); + + it('a rewound cursor makes the backlog pending again (measure AFTER rewind)', async () => { + // unleasedAtOrBelow measures against the lease cursor, so the watermark + // must be captured after rewind(): before it, the cursor still holds the + // previous connection's position and the backlog looks already-sent. + const q = new MemoryQueue('barrier-rewind'); + q.enqueue('a'); q.enqueue('b'); + await q.leaseNext(); await q.leaseNext(); // previous connection sent both + const w = await q.maxSeq(); + + expect(await q.unleasedAtOrBelow(w)).toBe(0); // stale cursor: looks clear + q.rewind(); // new connection resends + expect(await q.unleasedAtOrBelow(w)).toBe(2); // truth: both pending again + }); +}); diff --git a/tests/queueContract.js b/tests/queueContract.js new file mode 100644 index 0000000..f6bb12e --- /dev/null +++ b/tests/queueContract.js @@ -0,0 +1,63 @@ +import { describe, expect, it } from 'vitest'; + +/** The exact contract every outbox backend must satisfy. Backend-specific tests + * cover persistence and cross-context behavior separately. */ +export function queueContract (label, createQueue) { + describe(`${label} shared outbox contract`, () => { + it('leases in order without deleting', async () => { + const queue = createQueue('lease'); + queue.enqueue('a'); + queue.enqueue('b'); + expect(await queue.leaseNext()).toEqual({ seq: 1, item: 'a' }); + expect(await queue.leaseNext()).toEqual({ seq: 2, item: 'b' }); + expect(await queue.unconfirmedCount()).toBe(2); + }); + + it('confirms only explicitly named storage ids', async () => { + const queue = createQueue('confirm'); + for (const item of ['a', 'b', 'c']) queue.enqueue(item); + await queue.unconfirmedCount(); + queue.confirm([1, 3]); + expect(await queue.unconfirmedCount()).toBe(1); + queue.rewind(); + expect(await queue.leaseNext()).toEqual({ seq: 2, item: 'b' }); + }); + + it('rewind re-hands unconfirmed records', async () => { + const queue = createQueue('rewind'); + queue.enqueue('a'); + expect(await queue.leaseNext()).toEqual({ seq: 1, item: 'a' }); + queue.rewind(); + expect(await queue.leaseNext()).toEqual({ seq: 1, item: 'a' }); + }); + + it('a parked consumer wakes through the normal scan', async () => { + const queue = createQueue('park'); + const waiting = queue.leaseNext(); + queue.enqueue('later'); + expect(await waiting).toEqual({ seq: 1, item: 'later' }); + }); + + it('answers the watermark predicate and ignores later live traffic', async () => { + const queue = createQueue('barrier'); + queue.enqueue('backlog'); + const watermark = await queue.maxSeq(); + queue.enqueue('live'); + expect(await queue.unleasedAtOrBelow(watermark)).toBe(1); + await queue.leaseNext(); + expect(await queue.unleasedAtOrBelow(watermark)).toBe(0); + }); + + it('clear removes records without reusing their storage ids', async () => { + const queue = createQueue('clear'); + queue.enqueue('first'); + queue.enqueue('second'); + await queue.unconfirmedCount(); + queue.clear(); + await queue.unconfirmedCount(); + queue.enqueue('third'); + + expect(await queue.maxSeq()).toBeGreaterThan(2); + }); + }); +} diff --git a/tests/util.test.js b/tests/util.test.js index cc749ba..5f6d3e0 100644 --- a/tests/util.test.js +++ b/tests/util.test.js @@ -96,3 +96,47 @@ describe('util.js testing', () => { expect(util.copyFields(source, fields)).toEqual({ foo: 'bar' }); }); }); + +// ───────────────────────────────────────────────────────────────────────────── +// Event identity +// ───────────────────────────────────────────────────────────────────────────── +// +// eventId is what the ack protocol names, so these are protocol invariants +// rather than debugging conveniences: an ack that cannot be matched back to a +// record is an event that never gets deleted (resent forever) or, worse, the +// wrong record deleted. + +describe('event identity', () => { + it('stamps .., and the parts agree with the composite', () => { + const e = { event: 'ADD' }; + util.timestampEvent(e); + const m = e.metadata; + + expect(m.eventId).toBe(`${m.browserTag}.${m.sessionTag}.${m.sessionSeq}`); + expect(typeof m.sessionSeq).toBe('number'); + }); + + it('the sequence advances per event, and the session tag does not', () => { + const a = { event: 'A' }; const b = { event: 'B' }; + util.timestampEvent(a); + util.timestampEvent(b); + + expect(b.metadata.sessionSeq).toBe(a.metadata.sessionSeq + 1); + expect(b.metadata.sessionTag).toBe(a.metadata.sessionTag); + expect(b.metadata.eventId).not.toBe(a.metadata.eventId); + }); + + it('identity survives verboseEvents being off — it is not a debug extra', () => { + // Turning off verbose logging must not turn off the ack protocol's ability + // to name an event. Identity used to live inside this flag. + util.setVerboseEvents(false); + try { + const e = { event: 'QUIET' }; + util.timestampEvent(e); + expect(e.metadata.eventId).toBeTruthy(); + expect(e.metadata.human_ts).toBeUndefined(); // verbose extras gone + } finally { + util.setVerboseEvents(true); + } + }); +}); diff --git a/tests/websocketLogger.test.js b/tests/websocketLogger.test.js new file mode 100644 index 0000000..dc46ec6 --- /dev/null +++ b/tests/websocketLogger.test.js @@ -0,0 +1,338 @@ +import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'; +import { websocketLogger } from '../src/websocketLogger.js'; +import { QueueType } from '../src/queue.js'; +import { Queue as IndexedDBQueue } from '../src/indexeddbQueue.js'; +import { storage } from '../src/browserStorage.js'; +import 'fake-indexeddb/auto'; + +const NativeWebSocket = globalThis.WebSocket; + +class FakeWebSocket { + static instances = []; + static acknowledge = false; + static answerSnapshots = true; + static autoOpen = true; + + readyState = 0; + onopen = null; + onclose = null; + onerror = null; + onmessage = null; + sent = []; + sentRaw = []; + failSends = false; + deferClose = false; + + constructor () { + FakeWebSocket.instances.push(this); + if (FakeWebSocket.autoOpen) { + queueMicrotask(() => { + this.readyState = 1; + this.onopen?.(); + }); + } + } + + send (data) { + if (this.readyState !== 1) throw new Error('socket is not open'); + if (this.failSends) throw new Error('simulated send failure'); + const raw = String(data); + this.sentRaw.push(raw); + let frame; + try { + frame = JSON.parse(raw); + this.sent.push(frame); + } catch { + return; + } + if (frame.event === 'fetch_blob' && FakeWebSocket.answerSnapshots) { + queueMicrotask(() => this.onmessage?.({ + data: JSON.stringify({ status: 'fetch_blob', data: {} }) + })); + } else if (FakeWebSocket.acknowledge && frame.metadata?.eventId) { + queueMicrotask(() => this.onmessage?.({ + data: JSON.stringify({ status: 'ack', id: frame.metadata.eventId }) + })); + } + } + + receive (frame) { + this.onmessage?.({ data: JSON.stringify(frame) }); + } + + close () { + if (this.readyState === 3) return; + if (this.deferClose) { + this.readyState = 2; + return; + } + this.finishClose(); + } + + finishClose () { + this.readyState = 3; + this.onclose?.({}); + } +} + +beforeAll(() => { globalThis.WebSocket = FakeWebSocket; }); +afterAll(() => { globalThis.WebSocket = NativeWebSocket; }); +beforeEach(() => { + FakeWebSocket.acknowledge = false; + FakeWebSocket.answerSnapshots = true; + FakeWebSocket.autoOpen = true; + storage.set({ lo_server: undefined }); +}); + +async function connectedLogger (options = {}) { + const logger = websocketLogger('ws://test.invalid', { + namespace: crypto.randomUUID(), + queueType: QueueType.IN_MEMORY, + ...options + }); + await logger.init(); + await vi.waitFor(() => expect(FakeWebSocket.instances.at(-1)?.readyState).toBe(1)); + return { logger, socket: FakeWebSocket.instances.at(-1) }; +} + +describe('WebSocket adapter', () => { + it('durable mode retains records until their identity ack arrives', async () => { + const { logger, socket } = await connectedLogger({ fetchState: false }); + logger(JSON.stringify({ event: 'answer', value: 42 })); + await vi.waitFor(() => expect(socket.sent.some(frame => frame.event === 'answer')).toBe(true)); + expect(await logger.unackedCount()).toBe(1); + + const answer = socket.sent.find(frame => frame.event === 'answer'); + expect(answer.metadata.eventId).toBeTypeOf('string'); + socket.receive({ status: 'ack', id: answer.metadata.eventId }); + await vi.waitFor(async () => expect(await logger.unackedCount()).toBe(0)); + }); + + it('autoack confirms after an OPEN send and does not fetch state by default', async () => { + const { logger, socket } = await connectedLogger({ autoack: true }); + logger(JSON.stringify({ event: 'telemetry' })); + await vi.waitFor(() => expect(socket.sent.some(frame => frame.event === 'telemetry')).toBe(true)); + await vi.waitFor(async () => expect(await logger.unackedCount()).toBe(0)); + expect(socket.sent.some(frame => frame.event === 'fetch_blob')).toBe(false); + }); + + it('keeps fetch_blob outside the outbox and exposes a mid-session re-request', async () => { + const { logger, socket } = await connectedLogger({ fetchState: false }); + logger.requestState(); + await vi.waitFor(() => expect(socket.sent.filter(frame => frame.event === 'fetch_blob')).toHaveLength(1)); + expect(await logger.unackedCount()).toBe(0); + logger.requestState(); + await vi.waitFor(() => expect(socket.sent.filter(frame => frame.event === 'fetch_blob')).toHaveLength(2)); + }); + + it('drains a legacy stored record that cannot be named', async () => { + const namespace = crypto.randomUUID(); + const queue = new IndexedDBQueue(`lo-event:${encodeURIComponent(namespace)}:durable`); + queue.enqueue('{ legacy invalid json'); + await queue.unconfirmedCount(); + + const { logger, socket } = await connectedLogger({ + namespace, + queueType: QueueType.PERSISTENT, + fetchState: false + }); + await vi.waitFor(() => expect(socket.sentRaw).toContain('{ legacy invalid json')); + await vi.waitFor(async () => expect(await logger.unackedCount()).toBe(0)); + }); + + it('dispatches server side-channel frames', async () => { + const previousWindow = globalThis.window; + const eventTarget = new EventTarget(); + globalThis.window = eventTarget; + const received = {}; + for (const eventName of ['auth', 'save_blob_ack', 'save_blob_nack', 'server-event']) { + eventTarget.addEventListener(eventName, event => { received[eventName] = event.detail; }); + } + + try { + const { socket } = await connectedLogger({ fetchState: false }); + socket.receive({ status: 'auth', user_id: 'u-1', display_name: 'Ada' }); + socket.receive({ status: 'local_storage', key: 'server-key', value: 42 }); + socket.receive({ status: 'browser_event', event_type: 'server-event', detail: { ok: true } }); + socket.receive({ status: 'save_blob_ack', token: 7 }); + socket.receive({ status: 'save_blob_nack', token: 8 }); + + expect(received).toEqual({ + auth: { user_id: 'u-1', display_name: 'Ada' }, + 'server-event': { ok: true }, + save_blob_ack: { token: 7 }, + save_blob_nack: { token: 8 } + }); + const stored = await new Promise(resolve => { + storage.get(['user_id', 'display_name', 'server-key'], resolve); + }); + expect(stored).toEqual({ user_id: 'u-1', display_name: 'Ada', 'server-key': 42 }); + } finally { + if (previousWindow === undefined) delete globalThis.window; + else globalThis.window = previousWindow; + } + }); + + it('rejects application protocol frames at admission', async () => { + const { logger } = await connectedLogger({ fetchState: false }); + for (const event of ['fetch_blob', 'save_blob', 'lock_fields']) { + expect(() => logger(JSON.stringify({ event }))).toThrow(/reserved/); + } + }); + + it('ignores malformed server frames without stopping delivery', async () => { + const { logger, socket } = await connectedLogger({ fetchState: false }); + socket.onmessage({ data: 'not json' }); + logger(JSON.stringify({ event: 'after-malformed-frame' })); + + await vi.waitFor(() => { + expect(socket.sent.some(frame => frame.event === 'after-malformed-frame')).toBe(true); + }); + }); + + it('initializes only once when init is called repeatedly', async () => { + const before = FakeWebSocket.instances.length; + const logger = websocketLogger('ws://test.invalid', { + namespace: crypto.randomUUID(), + queueType: QueueType.IN_MEMORY, + fetchState: false + }); + + await Promise.all([logger.init(), logger.init()]); + await vi.waitFor(() => expect(FakeWebSocket.instances.length).toBe(before + 1)); + }); + + it('keeps a stable direct-use outbox namespace across a server override', async () => { + FakeWebSocket.autoOpen = false; + const originalServer = `ws://original-${crypto.randomUUID()}.invalid`; + storage.set({ lo_server: `ws://override-${crypto.randomUUID()}.invalid` }); + + const beforeInit = websocketLogger(originalServer, { + queueType: QueueType.PERSISTENT, + fetchState: false + }); + beforeInit(JSON.stringify({ event: 'before-init' })); + await vi.waitFor(async () => expect(await beforeInit.unackedCount()).toBe(1)); + + const afterInit = websocketLogger(originalServer, { + queueType: QueueType.PERSISTENT, + fetchState: false + }); + await afterInit.init(); + expect(await afterInit.unackedCount()).toBe(1); + }); + + it('sends a recovered backlog before requesting its snapshot', async () => { + FakeWebSocket.acknowledge = true; + const logger = websocketLogger('ws://test.invalid', { + namespace: crypto.randomUUID(), + queueType: QueueType.IN_MEMORY + }); + logger(JSON.stringify({ event: 'recovered-answer' })); + await logger.init(); + const socket = FakeWebSocket.instances.at(-1); + await vi.waitFor(() => expect(socket.sent.some(frame => frame.event === 'fetch_blob')).toBe(true)); + const events = socket.sent.map(frame => frame.event); + expect(events.indexOf('recovered-answer')).toBeLessThan(events.indexOf('fetch_blob')); + }); + + it('releases a lease parked on the previous connection before sending the new preamble', async () => { + const { logger, socket: first } = await connectedLogger({ fetchState: false }); + logger.setField(JSON.stringify({ + event: 'lock_fields', + fields: { source: 'test-app' }, + metadata: { eventId: 'initial-lock' } + })); + logger(JSON.stringify({ event: 'answer', metadata: { eventId: 'answer' } })); + await vi.waitFor(() => expect(first.sent.some(frame => frame.event === 'answer')).toBe(true)); + + first.close(); + await vi.waitFor( + () => expect(FakeWebSocket.instances.at(-1)).not.toBe(first), + { timeout: 4000 } + ); + const second = FakeWebSocket.instances.at(-1); + await vi.waitFor(() => expect(second.sent.some(frame => frame.event === 'answer')).toBe(true)); + + expect(second.sent.map(frame => frame.event)).toEqual([ + 'lock_fields', + 'answer', + 'lock_fields' + ]); + const lockIds = second.sent + .filter(frame => frame.event === 'lock_fields') + .map(frame => frame.metadata.eventId); + expect(lockIds[0]).toBe('initial-lock'); + expect(lockIds[1]).not.toBe('initial-lock'); + }); + + it('retires an OPEN socket whose send throws, then retries on a new connection', async () => { + const { logger, socket: first } = await connectedLogger({ fetchState: false }); + first.failSends = true; + logger(JSON.stringify({ event: 'answer', metadata: { eventId: 'answer' } })); + + await vi.waitFor(() => expect(first.readyState).toBe(3)); + expect(first.sent).toEqual([]); + expect(await logger.unackedCount()).toBe(1); + + await vi.waitFor( + () => expect(FakeWebSocket.instances.at(-1)).not.toBe(first), + { timeout: 4000 } + ); + const second = FakeWebSocket.instances.at(-1); + await vi.waitFor(() => expect(second.sent.some(frame => frame.event === 'answer')).toBe(true)); + }); + + it('reports a failed connection before its close event arrives', async () => { + const previousWindow = globalThis.window; + const eventTarget = new EventTarget(); + const statuses = []; + globalThis.window = eventTarget; + eventTarget.addEventListener('lo_connection_status', event => { + statuses.push(event.detail.connected); + }); + + try { + const { logger, socket } = await connectedLogger({ fetchState: false }); + await vi.waitFor(() => expect(statuses).toContain(true)); + socket.deferClose = true; + socket.failSends = true; + logger(JSON.stringify({ event: 'answer', metadata: { eventId: 'answer' } })); + + await vi.waitFor(() => expect(statuses.at(-1)).toBe(false)); + expect(socket.readyState).toBe(2); + expect(await logger.unackedCount()).toBe(1); + + socket.finishClose(); + } finally { + if (previousWindow === undefined) delete globalThis.window; + else globalThis.window = previousWindow; + } + }); + + // This mutates module-global disabler state permanently, so it stays last. + it('a permanent hold retains a parked record and a privacy opt-out clears it', async () => { + const { logger, socket } = await connectedLogger({ fetchState: false }); + // The drain loop is parked on an empty queue at this point. + socket.receive({ + status: 'blocklist', + message: 'hold', + time_limit: 'PERMANENT', + action: 'MAINTAIN' + }); + await new Promise(resolve => setTimeout(resolve, 0)); + logger(JSON.stringify({ event: 'must-stay-local' })); + + await vi.waitFor(async () => expect(await logger.unackedCount()).toBe(1)); + await new Promise(resolve => setTimeout(resolve, 30)); + expect(socket.sent.some(frame => frame.event === 'must-stay-local')).toBe(false); + + socket.receive({ + status: 'blocklist', + message: 'privacy opt-out', + time_limit: 'PERMANENT', + action: 'DROP' + }); + await vi.waitFor(async () => expect(await logger.unackedCount()).toBe(0)); + }); +}); diff --git a/tsup.config.ts b/tsup.config.ts index 31e2c4e..eafac2f 100644 --- a/tsup.config.ts +++ b/tsup.config.ts @@ -16,6 +16,7 @@ export default defineConfig({ memoryQueue: 'src/memoryQueue.ts', indexeddbQueue: 'src/indexeddbQueue.ts', types: 'src/types.ts', + hooks: 'src/hooks.ts', 'metadata/browserinfo': 'src/metadata/browserinfo.ts', 'metadata/chromeauth': 'src/metadata/chromeauth.ts', 'metadata/storage': 'src/metadata/storage.ts', @@ -24,7 +25,14 @@ export default defineConfig({ dts: true, sourcemap: true, clean: true, - splitting: false, + // splitting MUST stay true: several entry points (e.g. hooks.ts) re-export + // stateful module-level singletons from reduxLogger.ts (save status, the + // status-listener set, the redux store). With splitting:false, tsup inlines + // a SEPARATE copy of reduxLogger into each entry, so e.g. useSaved() in + // hooks.js reads a different _saveStatus than the store subscription in + // reduxLogger.js updates — the indicator gets stuck. Sharing a chunk keeps + // those singletons singular across entry points. + splitting: true, target: 'es2022', - external: ['ws', 'redux', 'redux-thunk', 'redux-state-sync', 'lodash'], + external: ['ws', 'redux', 'redux-thunk', 'redux-state-sync', 'lodash', 'react'], });