From 10673c71b402c04dc7608061beb6cf4388567395 Mon Sep 17 00:00:00 2001 From: Yos Riady Date: Mon, 10 Aug 2026 09:29:41 +0700 Subject: [PATCH 01/12] Port traffic-source sanitization and queue fixes from the web SDK MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Brings across three fixes that landed in getformo/sdk over the last two weeks and apply equally here, plus one bug found while verifying them. Sanitize traffic-source values (web #307) parseTrafficSource captured utm_* and ref verbatim, persisted them as sticky session attribution, and replayed them onto every subsequent event. On mobile the hostile input is not a scanner crawling a website: anyone who can get a user to open myapp://x?utm_source=")).toBe(""); + expect(sanitizeRef("javascript:alert(1)")).toBe(""); + expect(sanitizeRef("café")).toBe(""); + expect(sanitizeRef("code with spaces")).toBe(""); + }); + + it("drops values longer than 64 characters", () => { + expect(sanitizeRef("a".repeat(64))).toBe("a".repeat(64)); + expect(sanitizeRef("a".repeat(65))).toBe(""); + }); + + it("drops the empty string", () => { + expect(sanitizeRef("")).toBe(""); + }); + }); + + describe("sanitizeUtm", () => { + it("keeps free-form campaign values", () => { + expect(sanitizeUtm("spring sale 2026")).toBe("spring sale 2026"); + expect(sanitizeUtm("black+friday")).toBe("black+friday"); + expect(sanitizeUtm("Sommerkampagne für Schuhe")).toBe( + "Sommerkampagne für Schuhe" + ); + }); + + it("rejects markup and quote characters", () => { + expect(sanitizeUtm("")).toBe(""); + expect(sanitizeUtm(`"onload="alert(1)`)).toBe(""); + expect(sanitizeUtm("'\">()locxss")).toBe(""); + expect(sanitizeUtm("back\\slash")).toBe(""); + }); + + it("rejects dangerous scheme prefixes", () => { + expect(sanitizeUtm("javascript:alert(1)")).toBe(""); + expect(sanitizeUtm(" JavaScript:alert(1)")).toBe(""); + expect(sanitizeUtm("data:text/html;base64,PHN2Zz4=")).toBe(""); + expect(sanitizeUtm("vbscript:msgbox(1)")).toBe(""); + }); + + it("rejects control, zero-width and replacement characters", () => { + expect(sanitizeUtm("goo\u0000gle")).toBe(""); + expect(sanitizeUtm("goo\tgle")).toBe(""); + expect(sanitizeUtm("goo\ngle")).toBe(""); + expect(sanitizeUtm("goo\u200bgle")).toBe(""); + expect(sanitizeUtm("goo\u202egle")).toBe(""); + expect(sanitizeUtm("\ufeffgoogle")).toBe(""); + expect(sanitizeUtm("goo\ufffdgle")).toBe(""); + }); + + it("rejects absurdly long values", () => { + expect(sanitizeUtm("a".repeat(255))).toBe("a".repeat(255)); + expect(sanitizeUtm("a".repeat(256))).toBe(""); + }); + }); + + describe("sanitizeReferrer", () => { + it("keeps ordinary deep links and web URLs", () => { + expect(sanitizeReferrer("myapp://product?utm_source=twitter")).toBe( + "myapp://product?utm_source=twitter" + ); + expect(sanitizeReferrer("https://example.com/a/b?c=d&e=f")).toBe( + "https://example.com/a/b?c=d&e=f" + ); + }); + + it("drops URLs carrying an unencoded payload", () => { + expect( + sanitizeReferrer("myapp://x?utm_source=") + ).toBe(""); + expect(sanitizeReferrer("javascript:alert(1)")).toBe(""); + }); + + it("allows URLs longer than the UTM budget but bounds them at 2048", () => { + const long = `https://example.com/?q=${"a".repeat(1_000)}`; + expect(sanitizeReferrer(long)).toBe(long); + expect(sanitizeReferrer(`https://example.com/?q=${"a".repeat(2_048)}`)).toBe( + "" + ); + }); + }); + + describe("sanitizeTrafficSources", () => { + it("applies the right rule per field", () => { + expect( + sanitizeTrafficSources({ + utm_source: "spring sale", + utm_campaign: "", + ref: "friend123", + referrer: "myapp://home", + }) + ).toEqual({ + utm_source: "spring sale", + utm_campaign: "", + ref: "friend123", + referrer: "myapp://home", + }); + }); + + it("uses the strict token rule for ref, not the loose UTM rule", () => { + // A value with spaces passes the UTM rule but must fail as a ref. + expect(sanitizeUtm("spring sale")).toBe("spring sale"); + expect(sanitizeTrafficSources({ ref: "spring sale" }).ref).toBe(""); + }); + + it("leaves sparse objects and non-string values alone", () => { + expect(sanitizeTrafficSources({})).toEqual({}); + expect( + sanitizeTrafficSources({ utm_source: "" } as Record) + ).toEqual({ utm_source: "" }); + }); + }); + + describe("parseTrafficSource", () => { + it("drops a poisoned utm_source from a deep link", () => { + const ts = parseTrafficSource( + "myapp://product?utm_source=%3Cscript%3Ealert(1)%3C%2Fscript%3E&utm_medium=cpc" + ); + expect(ts.utm_source).toBe(""); + expect(ts.utm_medium).toBe("cpc"); + }); + + it("drops a poisoned ref from a deep link", () => { + const ts = parseTrafficSource( + "myapp://product?ref=javascript%3AdomxssExecutionSink(1)" + ); + expect(ts.ref).toBe(""); + }); + + it("drops a poisoned ref extracted via pathPattern", () => { + const ts = parseTrafficSource( + "https://example.com/invite/%3Cscript%3E", + undefined, + "^/invite/(.+)$" + ); + expect(ts.ref).toBe(""); + }); + + it("drops a poisoned ref from a custom query param", () => { + const ts = parseTrafficSource( + "myapp://x?partner_code=%3Cimg%20src%3Dx%20onerror%3Dalert(1)%3E", + ["partner_code"] + ); + expect(ts.ref).toBe(""); + }); + + it("keeps a clean deep link untouched", () => { + const ts = parseTrafficSource( + "myapp://product?utm_source=twitter&utm_medium=social&ref=friend123" + ); + expect(ts).toMatchObject({ + utm_source: "twitter", + utm_medium: "social", + ref: "friend123", + referrer: "myapp://product?utm_source=twitter&utm_medium=social&ref=friend123", + }); + }); + + it("sanitizes the referrer on the unparseable-URL fallback path", () => { + // No "?" at all, so the parser returns { referrer: url } directly. + const ts = parseTrafficSource("not a url "); + expect(ts.referrer).toBe(""); + }); + + it("sanitizes a poisoned Android install referrer query", () => { + // The shape produced by lib/installReferrer for a Play referrer string. + const ts = parseTrafficSource( + "https://play.google.com/store/apps?utm_source=%3Cscript%3E&utm_campaign=spring" + ); + expect(ts.utm_source).toBe(""); + expect(ts.utm_campaign).toBe("spring"); + }); + }); + + describe("stored values", () => { + it("flushes values poisoned by a pre-sanitization SDK on read", () => { + // Simulate storage written by an older build that did not sanitize. + storeTrafficSource({ + utm_source: "", + utm_campaign: "spring", + ref: "friend123", + }); + + const stored = getStoredTrafficSource(); + expect(stored?.utm_source).toBe(""); + expect(stored?.utm_campaign).toBe("spring"); + expect(stored?.ref).toBe("friend123"); + }); + + it("does not let a poisoned incoming value clobber a clean stored one", () => { + updateStoredTrafficSource({ utm_source: "twitter", ref: "friend123" }); + // A later deep link carrying a payload must not win the per-field merge. + updateStoredTrafficSource( + parseTrafficSource("myapp://x?utm_source=%3Cscript%3Ealert(1)%3C%2Fscript%3E") + ); + + const stored = getStoredTrafficSource(); + expect(stored?.utm_source).toBe("twitter"); + expect(stored?.ref).toBe("friend123"); + }); + }); +}); diff --git a/src/lib/event/EventQueue.ts b/src/lib/event/EventQueue.ts index f155b4d..a66199a 100644 --- a/src/lib/event/EventQueue.ts +++ b/src/lib/event/EventQueue.ts @@ -48,6 +48,19 @@ const MIN_FLUSH_INTERVAL = 1_000 * 10; // 10 seconds const noop = () => {}; +/** + * Invoke a consumer-supplied callback without letting it escape the SDK. + * These callbacks are arbitrary app code; a throw from one must not surface + * as an unhandled rejection from flush() or abort the remaining callbacks. + */ +const safeCall = (fn: (...args: unknown[]) => unknown, ...args: unknown[]) => { + try { + fn(...args); + } catch (error) { + logger.error("EventQueue: Callback threw, ignoring", error); + } +}; + /** * Event queue for React Native * Handles batching, flushing, and retries with app lifecycle awareness @@ -64,6 +77,15 @@ export class EventQueue implements IEventQueue { private payloadHashes: Set = new Set(); private flushMutex: Promise = Promise.resolve(); private appStateSubscription: { remove: () => void } | null = null; + /** + * Whether anything has been flushed yet this app session. Starts false so + * the first event is sent immediately (see enqueue). A cold start opened + * from an ad click or deep link produces its attribution events right away, + * and those are exactly the events lost if the process is killed before the + * batch timer fires or AppState reports background — a force-quit from the + * app switcher, an OS memory kill, or a crash never gives us that chance. + */ + private flushed = false; constructor(writeKey: string, options: Options) { this.writeKey = writeKey; @@ -189,12 +211,15 @@ export class EventQueue implements IEventQueue { 0 ) >= this.maxQueueSize; - if (hasReachedFlushAt || hasReachedQueueSize) { + // Ship the first event of the app session as a batch of one rather than + // holding it for flushAt/flushInterval; subsequent events keep batching. + if (hasReachedFlushAt || hasReachedQueueSize || !this.flushed) { // Clear timer to prevent double flush if (this.timer) { clearTimeout(this.timer); this.timer = null; } + this.flushed = true; // Flush uses internal mutex to serialize operations this.flush().catch((error) => { logger.error("EventQueue: Failed to flush on threshold", error); @@ -203,7 +228,16 @@ export class EventQueue implements IEventQueue { } if (this.flushIntervalMs && !this.timer) { - this.timer = setTimeout(this.flush.bind(this), this.flushIntervalMs); + // flush() rethrows once sendWithRetry is exhausted. Passing it to + // setTimeout bare left that rejection unhandled, surfacing in the host + // app as an "Uncaught (in promise)" on every failed interval flush — + // observed against a 4xx from the events API. The threshold and + // background paths already log and swallow; this one has to as well. + this.timer = setTimeout(() => { + this.flush().catch((error) => { + logger.error("EventQueue: Failed to flush on interval", error); + }); + }, this.flushIntervalMs); } } @@ -232,7 +266,7 @@ export class EventQueue implements IEventQueue { await previousMutex; if (!this.queue.length) { - callback(); + safeCall(callback); return; } @@ -246,9 +280,9 @@ export class EventQueue implements IEventQueue { const done = (err?: Error) => { items.forEach(({ message, callback: itemCallback }) => - itemCallback(err, message, data) + safeCall(itemCallback, err, message, data) ); - callback!(err, data); + safeCall(callback!, err, data); }; try { diff --git a/src/utils/sanitize.ts b/src/utils/sanitize.ts new file mode 100644 index 0000000..a222a07 --- /dev/null +++ b/src/utils/sanitize.ts @@ -0,0 +1,105 @@ +import type { ITrafficSource } from "../types"; + +/** + * Traffic-source value sanitization. + * + * Ported from the Formo web SDK, where vulnerability scanners (e.g. Acunetix) + * crawling customer sites injected XSS probes such as + * `javascript:domxssExecutionSink(1,"'\">()locxss")` or + * `` into every query parameter. Without validation + * those payloads are captured verbatim as utm_* / ref values, persisted as + * sticky session traffic sources, and pollute attribution reporting. + * + * React Native has the same exposure through two attacker-reachable inputs: + * + * - Deep links handed to `setTrafficSourceFromUrl`. Anyone who can get a user + * to open `myapp://x?utm_source=` controls the + * value verbatim, and it is persisted for the whole session. + * - The Android Play Install Referrer string, which is derived from the + * `referrer` parameter of a Play Store URL and is likewise attacker-supplied. + * + * Each field class gets the tightest rule its legitimate values allow: + * + * - Referral codes are short tokens; >99.5% of production values match the + * strict pattern and none of the remainder are legitimate (scanner + * payloads, mangled encodings, URLs glued to codes). + * - UTM values are free-form (spaces, unicode, `+` are legitimate), so they + * only reject markup/quote characters, dangerous URL schemes, control and + * zero-width characters, and absurd lengths. + * - `referrer` diverges from the web SDK, which leaves it untouched because + * there it is a browser-set `document.referrer` already handled by redactUrl. + * In React Native `referrer` holds the raw deep-link URL the attacker + * supplied, so sanitizing only the utm_ and ref fields would still let the + * payload through. It gets the same character rules with a URL-sized budget; + * a well-formed URL percent-encodes the rejected characters anyway. + * + * Invalid values are dropped to "" — the same representation as "parameter + * absent" — rather than repaired, so a poisoned value can never be persisted + * or reported. + * + * The web SDK additionally sanitizes ad-platform click IDs (gclid, fbclid, + * ...). This SDK does not capture them, so that rule is intentionally absent; + * add it here alongside the capture if click IDs are ever supported. + */ + +const REF_PATTERN = /^[A-Za-z0-9._-]{1,64}$/; + +const UTM_MAX_LENGTH = 255; + +// URLs are legitimately much longer than a UTM value, but not unbounded — +// this is well above any real deep link and still bounds what gets persisted. +const REFERRER_MAX_LENGTH = 2_048; + +// Markup/quote/backslash characters plus C0/C1 control characters and +// zero-width / bidi / BOM / replacement characters (mangled-encoding +// markers). Explicit ranges instead of \p{C} to avoid the `u`-flag +// property-escape requirement. +const FORBIDDEN_CHARS = + /[<>"'`\\\u0000-\u001f\u007f-\u009f\u200b-\u200f\u2028-\u202e\u2060\ufeff\ufffd]/; + +// Values smuggling an executable/URL scheme, e.g. `javascript:alert(1)`. +const FORBIDDEN_SCHEME_PREFIX = /^\s*(javascript|data|vbscript):/i; + +const sanitizeRef = (value: string): string => + REF_PATTERN.test(value) ? value : ""; + +const sanitizeUtm = (value: string): string => + value.length <= UTM_MAX_LENGTH && + !FORBIDDEN_CHARS.test(value) && + !FORBIDDEN_SCHEME_PREFIX.test(value) + ? value + : ""; + +const sanitizeReferrer = (value: string): string => + value.length <= REFERRER_MAX_LENGTH && + !FORBIDDEN_CHARS.test(value) && + !FORBIDDEN_SCHEME_PREFIX.test(value) + ? value + : ""; + +/** + * Sanitize every traffic-source field of a (possibly sparse) traffic-source + * object. Unknown keys fall through to the UTM rule, the most permissive of + * the value rules. + */ +export const sanitizeTrafficSources = >( + trafficSources: T +): T => { + const result: Record = { ...trafficSources }; + for (const key of Object.keys(result)) { + const value = result[key]; + if (typeof value !== "string" || value === "") { + continue; + } + if (key === "ref") { + result[key] = sanitizeRef(value); + } else if (key === "referrer") { + result[key] = sanitizeReferrer(value); + } else { + result[key] = sanitizeUtm(value); + } + } + return result as T; +}; + +export { sanitizeRef, sanitizeUtm, sanitizeReferrer }; diff --git a/src/utils/trafficSource.ts b/src/utils/trafficSource.ts index 9eb6536..479dac4 100644 --- a/src/utils/trafficSource.ts +++ b/src/utils/trafficSource.ts @@ -6,16 +6,30 @@ import { logger } from "../lib/logger"; import { storage } from "../lib/storage"; import { SESSION_TRAFFIC_SOURCE_KEY } from "../constants"; +import { sanitizeTrafficSources } from "./sanitize"; import type { ITrafficSource } from "../types"; /** * Parse UTM parameters and referral info from URL * Supports both web URLs (https://) and deep link URLs (myapp://) + * + * Every return path is sanitized (see ./sanitize) so scanner-injected or + * hand-crafted payloads in a deep link can never be persisted or reported. */ export function parseTrafficSource( url: string, customRefParams?: string[], pathPattern?: string +): Partial { + return sanitizeTrafficSources( + extractTrafficSource(url, customRefParams, pathPattern) + ); +} + +function extractTrafficSource( + url: string, + customRefParams?: string[], + pathPattern?: string ): Partial { try { // Handle deep link URLs that may not have standard URL format @@ -112,7 +126,11 @@ export function getStoredTrafficSource(): Partial | undefined { try { const stored = storage().get(SESSION_TRAFFIC_SOURCE_KEY); if (stored && typeof stored === "string") { - return JSON.parse(stored) as Partial; + // Sanitize on the way out too, so values persisted by a pre-sanitization + // SDK version are flushed rather than replayed onto every event. + return sanitizeTrafficSources( + JSON.parse(stored) as Partial + ); } } catch (error) { logger.debug("Failed to get stored traffic source:", error); From fc6724e9f0b4fcac6ef7bead7395216026f97ee1 Mon Sep 17 00:00:00 2001 From: Yos Riady Date: Mon, 10 Aug 2026 09:37:58 +0700 Subject: [PATCH 02/12] Ignore the two image-size advisories in the prod audit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit GHSA-w3rx-r6r6-pgpr and GHSA-5p2g-fcmc-qvqq started failing the audit job. Both name ">=2.0.3" as the patched range, but that version has never been published: npm's latest image-size is 2.0.2, which is itself inside the vulnerable "<=2.0.2" range. An overrides entry would be unresolvable, so there is no fix to take, breaking or otherwise. image-size reaches the graph only as metro > image-size (metro pins ^1.0.2, resolving 1.2.1) — the bundler's asset pipeline, which is build tooling and never ships in the SDK's runtime output. Both advisories are decode-time infinite loops on malformed ICNS/JXL/HEIF input, so reaching them would mean bundling a hostile image at build time. This is the same situation the existing @babel/core entry documents, and it follows the audit policy already recorded in this file: pin to a patched version where one exists, ignore build-tooling advisories that have no non-breaking fix. Scoped to these two GHSAs, so nothing else in the prod graph is masked — `pnpm audit --prod` now reports "2 high (2 ignored)" and exits 0. Co-Authored-By: Claude Opus 5 --- pnpm-workspace.yaml | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index 223b542..0ad8a57 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -85,9 +85,23 @@ overrides: # GHSA-4x5r-pxfx-6jf8 @babel/core (LOW) no patched version published (>=7.29.1 absent) # GHSA-96hv-2xvq-fx4p ws (HIGH) one build-tooling ws@8.20.1 resists the >=8.21.0 override # GHSA-h67p-54hq-rp68 js-yaml (MODERATE) the only 4.x fix is a breaking major bump (5.x) +# GHSA-w3rx-r6r6-pgpr image-size (HIGH) no patched version published (see below) +# GHSA-5p2g-fcmc-qvqq image-size (HIGH) no patched version published (see below) +# +# image-size: both advisories name ">=2.0.3" as the patched range, but 2.0.3 has +# never been published — npm's latest is 2.0.2, which is itself inside the +# vulnerable "<=2.0.2" range. An override is therefore unresolvable, and there is +# no fix to take, breaking or otherwise. It reaches the graph only as +# metro > image-size (metro pins ^1.0.2; 1.2.1 resolves), i.e. the asset pipeline +# of the bundler, which is build tooling and never ships in the SDK's runtime +# output. Both advisories are decode-time infinite loops on malformed ICNS/JXL/ +# HEIF input, so exposure would require bundling a hostile image at build time. +# Drop these two entries as soon as a patched image-size is published. # Revisit as upstream (react-native / metro / viem) updates these transitive deps. auditConfig: ignoreGhsas: - GHSA-4x5r-pxfx-6jf8 - GHSA-96hv-2xvq-fx4p - GHSA-h67p-54hq-rp68 + - GHSA-w3rx-r6r6-pgpr + - GHSA-5p2g-fcmc-qvqq From d23f15414644ed67d9b547aa505327ef6d06e49a Mon Sep 17 00:00:00 2001 From: Yos Riady Date: Mon, 10 Aug 2026 09:42:28 +0700 Subject: [PATCH 03/12] Re-arm the interval timer when a flush fails MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses Codex review feedback on #83. The first-event branch returns before the timer-arming block, and flush() clears the timer on entry. So when that immediate flush exhausted its retries, the items were unshifted back onto the queue with nothing scheduled to retry them: a foregrounded app that emitted no further event stranded them until the next enqueue or a background transition, neither of which is guaranteed. That is the worst case for this particular flush. Sending the first event immediately means it goes out during cold start, when the radio may still be waking, a VPN reconnecting, or a captive portal intercepting — so it is the attempt most likely to fail, carrying the attribution event the immediate flush exists to protect. The timer-arming logic moves into scheduleFlush(), which no-ops when there is nothing queued or a timer is already pending, and both failure paths now call it. Retries become self-sustaining: a queue that outlives a transient outage drains on the next interval once connectivity returns. The same gap existed for the flushAt/maxQueueSize thresholds before this branch was added; routing every session's first event through it is what made it likely rather than theoretical. Both new tests fail without the re-arm and pass with it. 333 pass (331 before), typecheck, lint and `pnpm audit --prod` clean. Co-Authored-By: Claude Opus 5 --- src/__tests__/eventQueue.test.ts | 70 ++++++++++++++++++++++++++++++++ src/lib/event/EventQueue.ts | 31 ++++++++++---- 2 files changed, 94 insertions(+), 7 deletions(-) diff --git a/src/__tests__/eventQueue.test.ts b/src/__tests__/eventQueue.test.ts index 7a2d22d..2568ed3 100644 --- a/src/__tests__/eventQueue.test.ts +++ b/src/__tests__/eventQueue.test.ts @@ -142,6 +142,76 @@ describe("EventQueue", () => { }); }); + describe("retry after a failed flush", () => { + it("re-arms the interval timer when the first-event flush fails", async () => { + jest.useFakeTimers(); + try { + fetchMock.mockResolvedValue({ ok: false, status: 500 }); + const queue = new EventQueue("test-write-key", { + apiHost: "https://events.formo.test", + flushAt: 20, + flushInterval: 10_000, + retryCount: 1, + }); + + // The first event flushes immediately. A cold start is exactly when the + // radio may still be waking, so this attempt is the one most likely to + // fail — and the event it carries is the attribution event. + await queue.enqueue(makeEvent(1)); + await jest.advanceTimersByTimeAsync(30_000); + + const afterFirstFlush = fetchMock.mock.calls.length; + expect(afterFirstFlush).toBeGreaterThan(0); + + // Nothing else is enqueued and the app stays foregrounded. The + // re-queued event must still get another attempt. + await jest.advanceTimersByTimeAsync(60_000); + expect(fetchMock.mock.calls.length).toBeGreaterThan(afterFirstFlush); + + queue.clear(); + await queue.cleanup(); + } finally { + jest.useRealTimers(); + } + }); + + it("keeps retrying until the send finally succeeds", async () => { + jest.useFakeTimers(); + try { + fetchMock.mockResolvedValue({ ok: false, status: 500 }); + const queue = new EventQueue("test-write-key", { + apiHost: "https://events.formo.test", + flushAt: 20, + flushInterval: 10_000, + retryCount: 1, + }); + + await queue.enqueue(makeEvent(1)); + await jest.advanceTimersByTimeAsync(60_000); + + // Only count what goes out AFTER connectivity returns — fetchMock also + // records the failed attempts, whose bodies carry the same event. + fetchMock.mockClear(); + fetchMock.mockResolvedValue({ ok: true, status: 200 }); + await jest.advanceTimersByTimeAsync(60_000); + + const delivered = fetchMock.mock.calls.flatMap( + ([, init]) => JSON.parse(init.body as string) as Array<{ event: string }> + ); + expect(delivered.map((e) => e.event)).toContain("event-1"); + + // And the queue is actually drained, not merely re-attempted. + fetchMock.mockClear(); + await jest.advanceTimersByTimeAsync(60_000); + expect(fetchMock).not.toHaveBeenCalled(); + + await queue.cleanup(); + } finally { + jest.useRealTimers(); + } + }); + }); + describe("consumer callbacks", () => { it("does not let a throwing callback escape flush() on an empty queue", async () => { const queue = makeQueue(); diff --git a/src/lib/event/EventQueue.ts b/src/lib/event/EventQueue.ts index a66199a..b0409cd 100644 --- a/src/lib/event/EventQueue.ts +++ b/src/lib/event/EventQueue.ts @@ -223,22 +223,39 @@ export class EventQueue implements IEventQueue { // Flush uses internal mutex to serialize operations this.flush().catch((error) => { logger.error("EventQueue: Failed to flush on threshold", error); + // A failed flush puts its items back on the queue, and this path + // returns without arming the interval timer. Re-arm, or a cold start + // whose immediate flush fails — the likeliest case, since the radio may + // still be waking — strands the attribution event until the next + // enqueue or a background transition, neither of which is guaranteed. + this.scheduleFlush(); }); return; } - if (this.flushIntervalMs && !this.timer) { + this.scheduleFlush(); + } + + /** + * Arm the batch-interval timer, if there is queued work and nothing is + * already scheduled. Safe to call repeatedly; it never stacks timers. + */ + private scheduleFlush(): void { + if (!this.flushIntervalMs || this.timer || !this.queue.length) return; + + this.timer = setTimeout(() => { // flush() rethrows once sendWithRetry is exhausted. Passing it to // setTimeout bare left that rejection unhandled, surfacing in the host // app as an "Uncaught (in promise)" on every failed interval flush — // observed against a 4xx from the events API. The threshold and // background paths already log and swallow; this one has to as well. - this.timer = setTimeout(() => { - this.flush().catch((error) => { - logger.error("EventQueue: Failed to flush on interval", error); - }); - }, this.flushIntervalMs); - } + this.flush().catch((error) => { + logger.error("EventQueue: Failed to flush on interval", error); + // Keep retrying on the interval so a queue that outlives a transient + // outage still drains once connectivity returns. + this.scheduleFlush(); + }); + }, this.flushIntervalMs); } /** From a9a6c270469826c433903d46a5c8b6c6435c743b Mon Sep 17 00:00:00 2001 From: Yos Riady Date: Mon, 10 Aug 2026 10:09:40 +0700 Subject: [PATCH 04/12] Stop retrying permanent failures, and close the queue on cleanup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses two Codex review findings on #83, both fallout from re-arming the interval timer in d23f154. Permanent send failures (P1) sendWithRetry deliberately does not retry a 4xx other than 429 — an invalid write key or a malformed batch — but flush() re-queued the items regardless, and the new re-arm then scheduled another attempt. The same rejected payload would be re-posted every interval for the lifetime of the process, re-invoking its callbacks and burning the user's battery and data each time. This was visible in the earlier examples-app run against the live API, where a placeholder write key returned 403 on every attempt. sendWithRetry now tags the error with whether another attempt could ever succeed, and flush() drops a batch marked retryable: false, releasing its payload hashes so equivalent events are not blocked from being enqueued later. Unexpected errors leave the flag undefined and are treated as retryable, so an unrecognised fault never silently discards events. Network errors and 5xx/429 keep the existing re-queue behaviour. This also moves closer to the web SDK, which does not re-queue a failed batch at all. Cleanup versus an in-flight flush (P2) flush() splices its items out of the queue before sending, so a cleanup that began while the immediate first-event flush was still in flight saw an empty queue, skipped its drain loop, and returned without waiting. The flush could then fail, re-queue, and arm a timer on an instance that had already been torn down — firing network calls after cleanup. cleanup() now sets a closed flag that scheduleFlush() honours, clears the timer up front, and awaits the flush mutex so an in-flight send settles before the drain loop measures what is left. Verified end-to-end against examples/with-react-native with a collector returning 400 on every request: three POSTs, each carrying distinct new events, then silence — the batches are dropped ("Dropping N event(s), permanently rejected by the API") rather than re-posted on a loop, and no unhandled rejection reaches the app. Each new test fails with its corresponding fix reverted. 336 pass (333 before), typecheck, lint and `pnpm audit --prod` clean. Co-Authored-By: Claude Opus 5 --- src/__tests__/eventQueue.test.ts | 103 +++++++++++++++++++++++++++++++ src/lib/event/EventQueue.ts | 68 ++++++++++++++++++-- 2 files changed, 165 insertions(+), 6 deletions(-) diff --git a/src/__tests__/eventQueue.test.ts b/src/__tests__/eventQueue.test.ts index 2568ed3..bc9b86a 100644 --- a/src/__tests__/eventQueue.test.ts +++ b/src/__tests__/eventQueue.test.ts @@ -212,6 +212,109 @@ describe("EventQueue", () => { }); }); + describe("permanent send failures", () => { + it("stops re-posting a batch the API rejected permanently", async () => { + jest.useFakeTimers(); + try { + // 400: an invalid write key or malformed payload. shouldRetry() is + // false, so sendWithRetry rejects without retrying. + fetchMock.mockResolvedValue({ ok: false, status: 400 }); + const queue = new EventQueue("test-write-key", { + apiHost: "https://events.formo.test", + flushAt: 20, + flushInterval: 10_000, + retryCount: 3, + }); + + await queue.enqueue(makeEvent(1)); + await jest.advanceTimersByTimeAsync(30_000); + + // One attempt, no retries — the status is not retryable. + expect(fetchMock).toHaveBeenCalledTimes(1); + + // And it must not be re-posted every interval for the process lifetime. + await jest.advanceTimersByTimeAsync(300_000); + expect(fetchMock).toHaveBeenCalledTimes(1); + + await queue.cleanup(); + } finally { + jest.useRealTimers(); + } + }); + + it("still retries a 429 rather than dropping it", async () => { + jest.useFakeTimers(); + try { + fetchMock.mockResolvedValue({ ok: false, status: 429 }); + const queue = new EventQueue("test-write-key", { + apiHost: "https://events.formo.test", + flushAt: 20, + flushInterval: 10_000, + retryCount: 1, + }); + + await queue.enqueue(makeEvent(1)); + await jest.advanceTimersByTimeAsync(30_000); + const afterFirst = fetchMock.mock.calls.length; + expect(afterFirst).toBeGreaterThan(1); // initial + retry + + await jest.advanceTimersByTimeAsync(60_000); + expect(fetchMock.mock.calls.length).toBeGreaterThan(afterFirst); + + queue.clear(); + await queue.cleanup(); + } finally { + jest.useRealTimers(); + } + }); + }); + + describe("cleanup", () => { + it("waits for an in-flight flush and arms no timer afterwards", async () => { + let releaseSend: (value: { ok: boolean; status: number }) => void; + const inFlight = new Promise<{ ok: boolean; status: number }>((resolve) => { + releaseSend = resolve; + }); + fetchMock.mockReturnValueOnce(inFlight); + + const queue = new EventQueue("test-write-key", { + apiHost: "https://events.formo.test", + flushAt: 20, + flushInterval: 10_000, + retryCount: 0, + }); + + // The first event flushes immediately; its items are spliced out of the + // queue, so a naive cleanup would see an empty queue and return early. + await queue.enqueue(makeEvent(1)); + await settle(); + expect(fetchMock).toHaveBeenCalledTimes(1); + + let cleanupDone = false; + const cleanup = queue.cleanup().then(() => { + cleanupDone = true; + }); + + await settle(); + expect(cleanupDone).toBe(false); // still waiting on the in-flight send + + // Fail it, so the items are re-queued during teardown. + releaseSend!({ ok: false, status: 500 }); + await cleanup; + expect(cleanupDone).toBe(true); + + // No timer may survive teardown and fire network calls afterwards. + const callsAtCleanup = fetchMock.mock.calls.length; + jest.useFakeTimers(); + try { + await jest.advanceTimersByTimeAsync(300_000); + expect(fetchMock.mock.calls.length).toBe(callsAtCleanup); + } finally { + jest.useRealTimers(); + } + }); + }); + describe("consumer callbacks", () => { it("does not let a throwing callback escape flush() on an empty queue", async () => { const queue = makeQueue(); diff --git a/src/lib/event/EventQueue.ts b/src/lib/event/EventQueue.ts index b0409cd..cedbcf5 100644 --- a/src/lib/event/EventQueue.ts +++ b/src/lib/event/EventQueue.ts @@ -22,6 +22,15 @@ type IFormoEventFlushPayload = IFormoEventPayload & { sent_at: string; }; +/** + * A send failure tagged with whether another attempt could ever succeed. + * `retryable: false` means the API rejected the payload itself (4xx other + * than 429) — re-posting the identical batch will fail identically forever. + * Left undefined for unexpected errors, which are treated as retryable so an + * unrecognised fault never silently discards events. + */ +type SendError = Error & { retryable?: boolean }; + interface Options { apiHost: string; flushAt?: number; @@ -86,6 +95,12 @@ export class EventQueue implements IEventQueue { * app switcher, an OS memory kill, or a crash never gives us that chance. */ private flushed = false; + /** + * Set once cleanup starts. A flush already in flight can fail and re-queue + * its items after teardown has begun; without this it would arm a timer on + * an instance that is going away, firing network calls post-cleanup. + */ + private closed = false; constructor(writeKey: string, options: Options) { this.writeKey = writeKey; @@ -241,6 +256,7 @@ export class EventQueue implements IEventQueue { * already scheduled. Safe to call repeatedly; it never stacks timers. */ private scheduleFlush(): void { + if (this.closed) return; if (!this.flushIntervalMs || this.timer || !this.queue.length) return; this.timer = setTimeout(() => { @@ -309,6 +325,21 @@ export class EventQueue implements IEventQueue { done(); logger.info(`Events sent successfully: ${data.length} events`); } catch (err) { + if ((err as SendError)?.retryable === false) { + // The API rejected this payload itself, so the identical batch can + // never succeed. Keeping it queued would re-post it every interval + // forever, re-invoking callbacks and burning the user's battery and + // data. Drop it, and release the hashes so equivalent events are not + // blocked from being enqueued again later. The web SDK likewise does + // not re-queue a failed batch. + items.forEach((item) => this.payloadHashes.delete(item.hash)); + done(err as Error); + logger.error( + `Dropping ${items.length} event(s), permanently rejected by the API:`, + err + ); + throw err; + } // Re-add items to the front of the queue for retry on next flush // Note: We intentionally keep hashes in payloadHashes to prevent duplicate // events from being enqueued while these items are pending retry. @@ -343,14 +374,25 @@ export class EventQueue implements IEventQueue { await new Promise((resolve) => setTimeout(() => resolve(), delay)); return this.sendWithRetry(data, attempt + 1); } - throw new Error(`HTTP error! status: ${response.status}`); + const error: SendError = new Error( + `HTTP error! status: ${response.status}` + ); + // A 4xx that is not 429 rejects this payload permanently — an invalid + // write key or a malformed batch. Tag it so flush() drops the batch + // instead of re-posting it on every interval for the process lifetime. + error.retryable = shouldRetry; + throw error; } } catch (error) { - if (isNetworkError(error) && attempt < this.retryCount) { - const delay = Math.pow(2, attempt) * 1000; - logger.warn(`Network error, retrying in ${delay}ms...`); - await new Promise((resolve) => setTimeout(() => resolve(), delay)); - return this.sendWithRetry(data, attempt + 1); + if (isNetworkError(error)) { + if (attempt < this.retryCount) { + const delay = Math.pow(2, attempt) * 1000; + logger.warn(`Network error, retrying in ${delay}ms...`); + await new Promise((resolve) => setTimeout(() => resolve(), delay)); + return this.sendWithRetry(data, attempt + 1); + } + // Connectivity comes back; keep these for a later attempt. + (error as SendError).retryable = true; } throw error; } @@ -385,6 +427,20 @@ export class EventQueue implements IEventQueue { * Clean up resources, flushing any pending events first */ public async cleanup(): Promise { + // Stop anything from arming a new timer for the rest of teardown. + this.closed = true; + if (this.timer) { + clearTimeout(this.timer); + this.timer = null; + } + + // A flush may already be in flight with its items spliced out of the + // queue, which would make the drain loop below see an empty queue and + // return before delivery finished. Wait for it to settle first — on a + // transient failure it puts those items back, and the loop then retries + // them. The mutex always resolves, including on a failed send. + await this.flushMutex; + // Flush all remaining queued events before teardown // Loop until queue is empty since flush() only sends flushAt events per call // Safety limit prevents infinite loops if flush silently fails From afe6cff2a5615ee68af741deabf9dfc73d820b54 Mon Sep 17 00:00:00 2001 From: Yos Riady Date: Mon, 10 Aug 2026 11:37:25 +0700 Subject: [PATCH 05/12] Harden the sanitizer and queue against the review findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Seven rounds of adversarial review over the queue and sanitizer. Each fix below has a regression test that was confirmed to fail with that fix — and only that fix — reverted. Sanitizer The referrer rule only caught RAW markup, but `Linking` and the browser percent-encode, so the realistic attack sailed through: it was close to a no-op. Values are now compared decoded, and the decoder had to get several things right before that was true. - Decode to a fixed point rather than a set number of layers. A fixed cap is reachable: re-encoding `javascript:alert(1)` grows its single `%` by two characters a layer, so 66 layers fit in 151 characters. The bound is now the input length, which is always sufficient because every productive pass turns three characters into one — that also proves termination. - Decode runs of consecutive escapes together. Byte-wise decoding can never reconstruct a multi-byte character, so `%E3%80%80javascript%3A...` never produced the U+3000 that lets the anchored scheme rule match. - Fall back to the largest decodable group when a run is not valid UTF-8. Otherwise a single `%C0` beside a `%3C` hid the `<` behind it. - Never let a malformed escape abort the decode. `decodeURIComponent` rejects the whole string on one stray `%`, which skipped every decoded check. - Test each separator-delimited segment for a dangerous scheme, not just the whole value. The rule is anchored, so an encoded `=` in `?utm_source%3Djavascript%253Aalert(1)` hid the payload mid-string. - Inspect query keys as well as values: a parameter with no `=` parses entirely as a key. - Apply the decoded checks to utm_* too, not only referrer. URLSearchParams strips one layer, so a twice-encoded payload arrived still encoded and passed a raw-only test — a bypass of the primary defence. The decoded form is checked against a narrower character set than the raw form, so an encoded quote in `?q=%22running%20shoes%22` is still kept. Queue - Do not resurrect events cleared mid-flush. flush() splices its batch out before sending, so clear() during opt-out could not see it and a later failure unshifted it back — delivering events after consent was withdrawn. A generation counter now invalidates those batches, and guards the success path too, where deleting hashes would have stripped the dedup entry of an identical event enqueued since. - Abort retries whose events were cleared during backoff. Retries span seconds, so opting out between attempts previously still posted the next. - Drop events enqueued across a clear() or cleanup(). enqueue() suspends on the async message-id hash; a caller that did not await it could resume after teardown and fire a request on a dead instance. - Ignore AppState transitions once cleanup starts, and detach the listener up front, so a backgrounding app cannot queue a flush that outlives teardown. - Re-arm from flush()'s own catch instead of each caller's, so every entry point is covered — the background flush cleared the interval on entry and left nothing scheduled, stranding its events. - Retry 408. It is transient, and dropping non-retryable statuses made it silent loss. - Catch rejected async callbacks. safeCall only caught synchronous throws. 363 tests pass (331 before), typecheck, lint and `pnpm audit --prod` clean. Two findings were deliberately not acted on, both recorded for follow-up: rejecting apostrophes and `javascript:`-prefixed campaign text is inherited verbatim from the web SDK, and diverging here alone would make the two SDKs report different values for the same campaign; and HTML entities / %uXXXX are not decoded by any URL consumer and do not yield executable markup, so covering them is an arms race with no stopping point. Co-Authored-By: Claude Opus 5 --- src/__tests__/eventQueue.test.ts | 297 ++++++++++++++++++ .../trafficSourceSanitization.test.ts | 160 ++++++++++ src/lib/event/EventQueue.ts | 156 +++++++-- src/utils/sanitize.ts | 199 +++++++++++- 4 files changed, 767 insertions(+), 45 deletions(-) diff --git a/src/__tests__/eventQueue.test.ts b/src/__tests__/eventQueue.test.ts index bc9b86a..2749b4a 100644 --- a/src/__tests__/eventQueue.test.ts +++ b/src/__tests__/eventQueue.test.ts @@ -315,7 +315,304 @@ describe("EventQueue", () => { }); }); + describe("clear() during an in-flight flush", () => { + it("does not resurrect events cleared while a batch was in flight", async () => { + let releaseSend: (value: { ok: boolean; status: number }) => void; + fetchMock.mockReturnValueOnce( + new Promise((resolve) => { + releaseSend = resolve; + }) + ); + + const queue = makeQueue(); + await queue.enqueue(makeEvent(1)); + await settle(); + expect(fetchMock).toHaveBeenCalledTimes(1); + + // Opt-out mid-flight. The batch is already spliced out of the queue, so + // clear() cannot see it. + queue.clear(); + + // The send then fails — those events must NOT come back. + releaseSend!({ ok: false, status: 500 }); + await settle(); + + const callsAfterClear = fetchMock.mock.calls.length; + jest.useFakeTimers(); + try { + await jest.advanceTimersByTimeAsync(300_000); + expect(fetchMock.mock.calls.length).toBe(callsAfterClear); + } finally { + jest.useRealTimers(); + } + + await queue.cleanup(); + }); + }); + + describe("background flush", () => { + it("re-arms the interval when the background flush fails", async () => { + const { AppState } = jest.requireMock("react-native") as { + AppState: { addEventListener: jest.Mock }; + }; + + jest.useFakeTimers(); + try { + // Succeed first, so the first-event flush drains the queue and leaves + // no timer pending — the background flush below is then the only thing + // that can clear one. + fetchMock.mockResolvedValue({ ok: true, status: 200 }); + const queue = new EventQueue("test-write-key", { + apiHost: "https://events.formo.test", + flushAt: 20, + flushInterval: 10_000, + // Must be truthy: the constructor treats 0 as "unset" and would + // substitute the default of 3, whose backoffs would then be + // indistinguishable from a re-armed interval flush. + retryCount: 1, + }); + + // The constructor registers the AppState listener; grab it. + const handler = AppState.addEventListener.mock.calls.at(-1)?.[1] as ( + s: string + ) => void; + expect(typeof handler).toBe("function"); + + await queue.enqueue(makeEvent(1)); + await jest.advanceTimersByTimeAsync(60_000); + + // Queue a second event: this arms the interval timer. + fetchMock.mockResolvedValue({ ok: false, status: 500 }); + await queue.enqueue(makeEvent(2)); + + // Background before that timer fires. flush() clears the timer on + // entry, and the send then fails and re-queues the event. + handler("background"); + // Long enough for both attempts and the 1s backoff to finish, but + // short of the 10s interval — so any later call can only come from a + // timer the failure path armed, not from sendWithRetry. + await jest.advanceTimersByTimeAsync(4_000); + const afterBackground = fetchMock.mock.calls.length; + expect(afterBackground).toBe(3); // 1 first-event + 2 background attempts + + // Nothing else happens — no new event, no foreground, no cleanup. The + // re-queued event must still get another attempt. One interval is + // enough to prove it; the flush re-arms on each failure, so advancing + // much further just spins the retry loop. + await jest.advanceTimersByTimeAsync(15_000); + expect(fetchMock.mock.calls.length).toBeGreaterThan(afterBackground); + + // Deliberately no cleanup() here. The re-armed flush is mid retry + // backoff, sleeping in a fake timer; cleanup() awaits the flush mutex, + // which that flush only releases once its backoff fires. Advancing far + // enough just re-arms again, so awaiting cleanup would hang. Dropping + // the fake timers below discards the pending work instead. + queue.clear(); + } finally { + jest.useRealTimers(); + } + }); + }); + + describe("background transition during teardown", () => { + it("ignores a background flush once cleanup has started", async () => { + const { AppState } = jest.requireMock("react-native") as { + AppState: { addEventListener: jest.Mock }; + }; + + jest.useFakeTimers(); + try { + fetchMock.mockResolvedValue({ ok: false, status: 500 }); + const queue = new EventQueue("test-write-key", { + apiHost: "https://events.formo.test", + flushAt: 20, + flushInterval: 10_000, + retryCount: 1, + }); + const handler = AppState.addEventListener.mock.calls.at(-1)?.[1] as ( + s: string + ) => void; + + // Fail the send so the event survives in the queue through teardown. + await queue.enqueue(makeEvent(1)); + await jest.advanceTimersByTimeAsync(5_000); + + const cleanupPromise = queue.cleanup(); + await jest.advanceTimersByTimeAsync(30_000); + await cleanupPromise; + + // A backgrounding app after teardown must not start another send. + fetchMock.mockClear(); + handler("background"); + await jest.advanceTimersByTimeAsync(30_000); + expect(fetchMock).not.toHaveBeenCalled(); + } finally { + jest.useRealTimers(); + } + }); + }); + + describe("clear() during retry backoff", () => { + it("abandons a batch cleared between retry attempts", async () => { + jest.useFakeTimers(); + try { + // Retryable, so the first failure schedules a backoff before attempt 2. + fetchMock.mockResolvedValue({ ok: false, status: 500 }); + const queue = new EventQueue("test-write-key", { + apiHost: "https://events.formo.test", + flushAt: 20, + flushInterval: 10_000, + retryCount: 3, + }); + + await queue.enqueue(makeEvent(1)); + // Let attempt 1 fail and enter backoff, but not reach attempt 2. + await jest.advanceTimersByTimeAsync(100); + const callsBeforeOptOut = fetchMock.mock.calls.length; + expect(callsBeforeOptOut).toBe(1); + + // Consent withdrawn mid-backoff. + queue.clear(); + + // Even if the API would now accept them, no further attempt may go out. + fetchMock.mockResolvedValue({ ok: true, status: 200 }); + await jest.advanceTimersByTimeAsync(300_000); + expect(fetchMock.mock.calls.length).toBe(callsBeforeOptOut); + + await queue.cleanup(); + } finally { + jest.useRealTimers(); + } + }); + }); + + describe("dedup across clear()", () => { + it("does not strip the dedup entry of an event re-enqueued after clear()", async () => { + let releaseSend: (value: { ok: boolean; status: number }) => void; + fetchMock.mockReturnValueOnce( + new Promise((resolve) => { + releaseSend = resolve; + }) + ); + + const queue = makeQueue({ flushAt: 20 }); + await queue.enqueue(makeEvent(1)); + await settle(); + expect(fetchMock).toHaveBeenCalledTimes(1); + + // Opt out, then the same event is produced again and re-enqueued. + queue.clear(); + await queue.enqueue(makeEvent(1)); + await settle(); + + // The original send now succeeds. It must not delete the hash that the + // newly queued copy of the same event depends on. + releaseSend!({ ok: true, status: 200 }); + await settle(); + + // A third identical enqueue must still be recognised as a duplicate. + fetchMock.mockClear(); + fetchMock.mockResolvedValue({ ok: true, status: 200 }); + await queue.enqueue(makeEvent(1)); + await queue.flush(); + await settle(); + + const delivered = fetchMock.mock.calls.flatMap( + ([, init]) => JSON.parse(init.body as string) as Array<{ event: string }> + ); + expect(delivered.filter((e) => e.event === "event-1")).toHaveLength(1); + + await queue.cleanup(); + }); + }); + + describe("enqueue racing clear()", () => { + it("drops an event whose hashing was still pending when the user opted out", async () => { + const queue = makeQueue(); + + // Do NOT await: enqueue suspends on the async message-id hash, so the + // event has not reached the queue that clear() empties. + const pending = queue.enqueue(makeEvent(1)); + queue.clear(); + await pending; + await settle(); + + expect(fetchMock).not.toHaveBeenCalled(); + + await queue.cleanup(); + }); + }); + + describe("enqueue racing cleanup", () => { + it("drops an event whose hashing was still pending when cleanup ran", async () => { + const queue = makeQueue(); + + // Do NOT await: enqueue suspends on the async message-id hash. + const pending = queue.enqueue(makeEvent(1)); + await queue.cleanup(); + await pending; + await settle(); + + // Nothing may be sent on an instance that has already been torn down. + expect(fetchMock).not.toHaveBeenCalled(); + }); + }); + + describe("retryable status codes", () => { + it("retries a 408 request timeout instead of dropping the batch", async () => { + jest.useFakeTimers(); + try { + fetchMock.mockResolvedValue({ ok: false, status: 408 }); + const queue = new EventQueue("test-write-key", { + apiHost: "https://events.formo.test", + flushAt: 20, + flushInterval: 10_000, + retryCount: 1, + }); + + await queue.enqueue(makeEvent(1)); + await jest.advanceTimersByTimeAsync(30_000); + const afterFirst = fetchMock.mock.calls.length; + expect(afterFirst).toBeGreaterThan(1); // retried, not dropped + + // Still queued, so the interval keeps trying. + await jest.advanceTimersByTimeAsync(60_000); + expect(fetchMock.mock.calls.length).toBeGreaterThan(afterFirst); + + queue.clear(); + await queue.cleanup(); + } finally { + jest.useRealTimers(); + } + }); + }); + describe("consumer callbacks", () => { + it("does not let a rejected async callback escape as an unhandled rejection", async () => { + const unhandled: unknown[] = []; + const onUnhandled = (reason: unknown) => unhandled.push(reason); + const proc = (globalThis as unknown as { process: NodeProcess }).process; + proc.on("unhandledRejection", onUnhandled); + + try { + const queue = makeQueue(); + // An async callback reports failure by rejecting, not by throwing. + const rejecting = async () => { + throw new Error("async callback blew up"); + }; + + await queue.enqueue(makeEvent(1), rejecting); + await settle(); + await queue.flush(rejecting); + await settle(); + + expect(unhandled).toEqual([]); + await queue.cleanup(); + } finally { + proc.off("unhandledRejection", onUnhandled); + } + }); + it("does not let a throwing callback escape flush() on an empty queue", async () => { const queue = makeQueue(); const throwing = () => { diff --git a/src/__tests__/trafficSourceSanitization.test.ts b/src/__tests__/trafficSourceSanitization.test.ts index 484e85a..7c49fe0 100644 --- a/src/__tests__/trafficSourceSanitization.test.ts +++ b/src/__tests__/trafficSourceSanitization.test.ts @@ -104,6 +104,41 @@ describe("traffic source sanitization", () => { expect(sanitizeUtm("goo\ufffdgle")).toBe(""); }); + it("rejects a double-encoded payload that survives one decode", () => { + // URLSearchParams decodes one layer before the value reaches sanitizeUtm, + // so a twice-encoded payload arrives as this literal — no raw markup. + expect(sanitizeUtm("%3Cscript%3Ealert(1)%3C%2Fscript%3E")).toBe(""); + expect(sanitizeUtm("%253Cscript%253E")).toBe(""); + expect(sanitizeUtm("javascript%3Aalert(1)")).toBe(""); + }); + + it("rejects a scheme hidden behind an encoded multi-byte space", () => { + // %E3%80%80 is U+3000 IDEOGRAPHIC SPACE, which "^\\s*javascript:" is meant + // to skip over. It only decodes when the whole escape run is decoded + // together — %E3 on its own is not valid UTF-8. + expect(sanitizeUtm("%E3%80%80javascript%3Aalert(1)")).toBe(""); + expect( + sanitizeReferrer("myapp://x?utm_source=%25E3%2580%2580javascript%253Aalert(1)") + ).toBe(""); + }); + + it("rejects a scheme hidden behind an invalid UTF-8 byte", () => { + // %C0 is not valid UTF-8. Decoding the run as a unit throws, so a naive + // decoder returns it whole and the "<" never materializes. + expect(sanitizeUtm("%3C%C0img src=x onerror=alert(1)")).toBe(""); + expect(sanitizeUtm("%C0%3Cscript%3E")).toBe(""); + }); + + it("keeps legitimate encoded multi-byte text", () => { + // café — decodes cleanly and contains nothing forbidden. + expect(sanitizeUtm("caf%C3%A9")).toBe("caf%C3%A9"); + }); + + it("keeps a percent sign that is not an escape sequence", () => { + expect(sanitizeUtm("50% off")).toBe("50% off"); + expect(sanitizeUtm("100%")).toBe("100%"); + }); + it("rejects absurdly long values", () => { expect(sanitizeUtm("a".repeat(255))).toBe("a".repeat(255)); expect(sanitizeUtm("a".repeat(256))).toBe(""); @@ -127,6 +162,122 @@ describe("traffic source sanitization", () => { expect(sanitizeReferrer("javascript:alert(1)")).toBe(""); }); + it("drops a payload that is percent-encoded, as a real URL carries it", () => { + // The raw string has no "<" at all — Linking and the browser encode it — + // so a raw-only check would pass this straight through. + expect( + sanitizeReferrer("myapp://x?utm_source=%3Cscript%3Ealert(1)%3C%2Fscript%3E") + ).toBe(""); + expect(sanitizeReferrer("myapp://x?ref=%3Cimg%20src%3Dx%20onerror%3D1%3E")).toBe( + "" + ); + expect(sanitizeReferrer("%6Aavascript:alert(1)")).toBe(""); + }); + + it("drops a double-encoded payload", () => { + expect( + sanitizeReferrer("myapp://x?utm_source=%253Cscript%253Ealert(1)%253C%252Fscript%253E") + ).toBe(""); + }); + + it("keeps encoded characters that are legitimate in a query value", () => { + // Quotes and spaces decode to harmless text and are common in search + // deep links, so the decoded check must not reject them. + expect( + sanitizeReferrer("myapp://search?q=%22running%20shoes%22") + ).toBe("myapp://search?q=%22running%20shoes%22"); + expect(sanitizeReferrer("myapp://x?name=T-Shirt%20%26%20Jeans")).toBe( + "myapp://x?name=T-Shirt%20%26%20Jeans" + ); + }); + + it("survives a malformed percent escape rather than throwing", () => { + // decodeURIComponent throws on a stray "%"; the value must still be + // evaluated, not crash traffic-source capture. + expect(() => sanitizeReferrer("myapp://x?q=100%")).not.toThrow(); + expect(sanitizeReferrer("myapp://x?q=100%")).toBe("myapp://x?q=100%"); + }); + + it("is not bypassed by appending a malformed escape to a payload", () => { + // A stray trailing "%" makes decodeURIComponent reject the whole string, + // which would otherwise skip every decoded check. + expect( + sanitizeReferrer("myapp://x?utm_source=%3Cscript%3Ealert(1)%3C%2Fscript%3E%") + ).toBe(""); + }); + + it("drops a dangerous scheme smuggled inside a query value", () => { + // The URL itself starts with "myapp://", so an anchored check against the + // whole string never matches; the parameter has to be read on its own. + expect(sanitizeReferrer("myapp://x?utm_source=javascript%3Aalert(1)")).toBe( + "" + ); + expect(sanitizeReferrer("myapp://x?r=data%3Atext%2Fhtml%3Bbase64%2CPHN2Zz4=")).toBe( + "" + ); + }); + + it("drops a deeply multi-encoded payload", () => { + expect( + sanitizeReferrer("myapp://x?utm_source=%2525253Cscript%2525253E") + ).toBe(""); + }); + + it("drops a payload encoded past any fixed layer limit", () => { + // Decoding runs to a fixed point rather than a set number of layers. + expect( + sanitizeReferrer( + "myapp://x?utm_source=%252525252525253Cscript%252525252525253E" + ) + ).toBe(""); + const deep = `myapp://x?u=${"%25".repeat(20)}3Cscript${"%25".repeat(20)}3E`; + expect(sanitizeReferrer(deep)).toBe(""); + }); + + it("drops a scheme re-encoded past any small constant layer bound", () => { + // Re-encoding "javascript:alert(1)" only grows its single "%" by two + // characters a layer, so 66 layers fit in ~151 characters — well under + // any plausible fixed cap. + let payload = "javascript:alert(1)"; + for (let i = 0; i < 66; i++) payload = encodeURIComponent(payload); + expect(payload.length).toBeLessThan(200); + expect(sanitizeUtm(payload)).toBe(""); + expect(sanitizeReferrer(`myapp://x?utm_source=${payload}`)).toBe(""); + }); + + it("drops a scheme hidden behind an encoded separator", () => { + // The "=" is encoded, so URLSearchParams reports a single key that only + // becomes "utm_source=javascript:alert(1)" after decoding — at which + // point an anchored scheme test no longer sees the payload at the start. + expect( + sanitizeReferrer("myapp://x?utm_source%3Djavascript%253Aalert(1)") + ).toBe(""); + expect( + sanitizeReferrer( + "utm_source%3Djavascript%253Aalert(1)%26utm_medium%3Dcpc" + ) + ).toBe(""); + }); + + it("drops a dangerous scheme hidden in a query KEY with no value", () => { + // "?%6Aavascript:alert(1)" has no "=", so it parses entirely as a key + // and contributes no value to inspect. + expect( + sanitizeReferrer("https://play.google.com/store/apps?%6Aavascript:alert(1)") + ).toBe(""); + expect(sanitizeReferrer("myapp://x?javascript:alert(1)")).toBe(""); + }); + + it("sanitizes a bare Android install-referrer query string", () => { + // No "?" and no scheme — just the raw referrer parameter from Play. + expect( + sanitizeReferrer("utm_source=%3Cscript%3E&utm_medium=cpc") + ).toBe(""); + expect(sanitizeReferrer("utm_source=google&utm_medium=cpc")).toBe( + "utm_source=google&utm_medium=cpc" + ); + }); + it("allows URLs longer than the UTM budget but bounds them at 2048", () => { const long = `https://example.com/?q=${"a".repeat(1_000)}`; expect(sanitizeReferrer(long)).toBe(long); @@ -176,6 +327,15 @@ describe("traffic source sanitization", () => { expect(ts.utm_medium).toBe("cpc"); }); + it("drops a double-encoded utm payload end to end", () => { + const ts = parseTrafficSource( + "myapp://product?utm_source=%253Cscript%253Ealert(1)%253C%252Fscript%253E&utm_medium=cpc" + ); + expect(ts.utm_source).toBe(""); + expect(ts.utm_medium).toBe("cpc"); + expect(JSON.stringify(ts)).not.toContain("script"); + }); + it("drops a poisoned ref from a deep link", () => { const ts = parseTrafficSource( "myapp://product?ref=javascript%3AdomxssExecutionSink(1)" diff --git a/src/lib/event/EventQueue.ts b/src/lib/event/EventQueue.ts index cedbcf5..a3d591d 100644 --- a/src/lib/event/EventQueue.ts +++ b/src/lib/event/EventQueue.ts @@ -64,7 +64,14 @@ const noop = () => {}; */ const safeCall = (fn: (...args: unknown[]) => unknown, ...args: unknown[]) => { try { - fn(...args); + const result = fn(...args); + // An `async` callback signals failure by returning a rejected promise + // rather than throwing, which the catch below would never see. + if (result && typeof (result as PromiseLike).then === "function") { + Promise.resolve(result).catch((error) => { + logger.error("EventQueue: Async callback rejected, ignoring", error); + }); + } } catch (error) { logger.error("EventQueue: Callback threw, ignoring", error); } @@ -101,6 +108,13 @@ export class EventQueue implements IEventQueue { * an instance that is going away, firing network calls post-cleanup. */ private closed = false; + /** + * Bumped by clear(). flush() splices its batch out of the queue before + * sending, so a clear() during opt-out cannot see those items; without this + * a later send failure would unshift them back and they would be delivered + * after consent was withdrawn. + */ + private generation = 0; constructor(writeKey: string, options: Options) { this.writeKey = writeKey; @@ -144,6 +158,11 @@ export class EventQueue implements IEventQueue { * Handle app state changes */ private handleAppStateChange(nextAppState: AppStateStatus): void { + // Teardown is already draining the queue. A flush queued here would wait + // on the mutex and could run — and re-queue on failure — after cleanup() + // has returned, leaving network work with no owner. + if (this.closed) return; + // Flush when app goes to background or becomes inactive if (nextAppState === "background" || nextAppState === "inactive") { logger.debug("EventQueue: App going to background, flushing events"); @@ -181,10 +200,33 @@ export class EventQueue implements IEventQueue { event: IFormoEvent, callback?: (...args: unknown[]) => void ): Promise { + if (this.closed) { + logger.debug("EventQueue: Ignoring event enqueued after cleanup"); + return; + } + callback = callback || noop; + const generation = this.generation; const message_id = await this.generateMessageId(event); + // Hashing is async, so cleanup() can complete while this call is suspended + // above. Re-check, or a caller that did not await enqueue() would resume + // after teardown, push onto a queue nobody will drain, and — if this is the + // session's first event — fire a network request on a torn-down instance. + if (this.closed) { + logger.debug("EventQueue: Ignoring event enqueued after cleanup"); + return; + } + + // Same window, but for opt-out: clear() cannot see an event that has not + // reached the queue yet, so an enqueue suspended across it would land + // afterwards and be delivered despite consent having been withdrawn. + if (this.generation !== generation) { + logger.debug("EventQueue: Ignoring event enqueued before opt-out"); + return; + } + // Check for duplicate if (this.isDuplicate(message_id)) { logger.warn( @@ -236,14 +278,11 @@ export class EventQueue implements IEventQueue { } this.flushed = true; // Flush uses internal mutex to serialize operations + // A failed flush re-queues its items and re-arms the interval itself, so + // a cold start whose immediate flush fails — the likeliest case, since + // the radio may still be waking — still retries the attribution event. this.flush().catch((error) => { logger.error("EventQueue: Failed to flush on threshold", error); - // A failed flush puts its items back on the queue, and this path - // returns without arming the interval timer. Re-arm, or a cold start - // whose immediate flush fails — the likeliest case, since the radio may - // still be waking — strands the attribution event until the next - // enqueue or a background transition, neither of which is guaranteed. - this.scheduleFlush(); }); return; } @@ -265,11 +304,10 @@ export class EventQueue implements IEventQueue { // app as an "Uncaught (in promise)" on every failed interval flush — // observed against a 4xx from the events API. The threshold and // background paths already log and swallow; this one has to as well. + // flush() re-arms on failure, so a queue that outlives a transient + // outage keeps retrying and drains once connectivity returns. this.flush().catch((error) => { logger.error("EventQueue: Failed to flush on interval", error); - // Keep retrying on the interval so a queue that outlives a transient - // outage still drains once connectivity returns. - this.scheduleFlush(); }); }, this.flushIntervalMs); } @@ -304,6 +342,9 @@ export class EventQueue implements IEventQueue { } const items = this.queue.splice(0, this.flushAt); + // Snapshot after the splice: from here on these items live only in this + // closure, so a clear() cannot reach them and we have to detect it. + const generation = this.generation; const sentAt = new Date().toISOString(); const data: IFormoEventFlushPayload[] = items.map((item) => ({ @@ -319,13 +360,28 @@ export class EventQueue implements IEventQueue { }; try { - await this.sendWithRetry(data); - // Only remove hashes after successful send - items.forEach((item) => this.payloadHashes.delete(item.hash)); + await this.sendWithRetry(data, generation); + // Only remove hashes after successful send, and only if clear() has + // not run meanwhile: it already emptied the set, so an identical event + // may have been enqueued since and now owns that hash. Deleting it + // here would strip the new item's dedup entry and let a duplicate + // through. + if (this.generation === generation) { + items.forEach((item) => this.payloadHashes.delete(item.hash)); + } done(); logger.info(`Events sent successfully: ${data.length} events`); } catch (err) { - if ((err as SendError)?.retryable === false) { + if (this.generation !== generation) { + // clear() ran while this batch was in flight — the consumer opted + // out. Putting these back would deliver events after consent was + // withdrawn, and clear() already emptied payloadHashes, so a + // resurrected item would also no longer be deduped. + done(err as Error); + logger.debug( + `EventQueue: Discarding ${items.length} in-flight event(s) cleared mid-flush` + ); + } else if ((err as SendError)?.retryable === false) { // The API rejected this payload itself, so the identical batch can // never succeed. Keeping it queued would re-post it every interval // forever, re-invoking callbacks and burning the user's battery and @@ -338,14 +394,21 @@ export class EventQueue implements IEventQueue { `Dropping ${items.length} event(s), permanently rejected by the API:`, err ); - throw err; + } else { + // Re-add items to the front of the queue for retry on next flush + // Note: We intentionally keep hashes in payloadHashes to prevent duplicate + // events from being enqueued while these items are pending retry. + this.queue.unshift(...items); + done(err as Error); + logger.error("Error sending events, re-queued for retry:", err); } - // Re-add items to the front of the queue for retry on next flush - // Note: We intentionally keep hashes in payloadHashes to prevent duplicate - // events from being enqueued while these items are pending retry. - this.queue.unshift(...items); - done(err as Error); - logger.error("Error sending events, re-queued for retry:", err); + + // Re-arm here rather than in each caller's catch, so EVERY entry point + // is covered — the background AppState flush and a consumer's manual + // flush() included. flush() clears the timer on entry, so without this + // a failed background flush would re-queue its items and leave nothing + // scheduled to retry them. No-ops when the queue is empty or closed. + this.scheduleFlush(); throw err; } } finally { @@ -353,11 +416,28 @@ export class EventQueue implements IEventQueue { } } + /** + * Abort a batch whose events were cleared while it sat in retry backoff. + * Retries span seconds, so a consumer can opt out between attempts; posting + * the next one would deliver events after consent was withdrawn. + */ + private assertNotCleared(generation: number): void { + if (this.generation !== generation) { + const error: SendError = new Error( + "EventQueue: batch cleared during retry backoff" + ); + // Never re-queue: these events were explicitly discarded. + error.retryable = false; + throw error; + } + } + /** * Send events with retry logic */ private async sendWithRetry( data: IFormoEventFlushPayload[], + generation: number, attempt = 0 ): Promise { try { @@ -372,7 +452,8 @@ export class EventQueue implements IEventQueue { if (shouldRetry && attempt < this.retryCount) { const delay = Math.pow(2, attempt) * 1000; await new Promise((resolve) => setTimeout(() => resolve(), delay)); - return this.sendWithRetry(data, attempt + 1); + this.assertNotCleared(generation); + return this.sendWithRetry(data, generation, attempt + 1); } const error: SendError = new Error( `HTTP error! status: ${response.status}` @@ -389,7 +470,8 @@ export class EventQueue implements IEventQueue { const delay = Math.pow(2, attempt) * 1000; logger.warn(`Network error, retrying in ${delay}ms...`); await new Promise((resolve) => setTimeout(() => resolve(), delay)); - return this.sendWithRetry(data, attempt + 1); + this.assertNotCleared(generation); + return this.sendWithRetry(data, generation, attempt + 1); } // Connectivity comes back; keep these for a later attempt. (error as SendError).retryable = true; @@ -402,8 +484,11 @@ export class EventQueue implements IEventQueue { * Check if error should be retried */ private shouldRetry(status: number): boolean { - // Retry on server errors (5xx) and rate limiting (429) - return (status >= 500 && status <= 599) || status === 429; + // Retry on server errors (5xx), rate limiting (429) and request timeout + // (408). 408 matters now that a non-retryable status drops the batch: a + // proxy or server timing out a request is transient, and treating it as + // permanent would silently lose those events. + return (status >= 500 && status <= 599) || status === 429 || status === 408; } /** @@ -412,6 +497,9 @@ export class EventQueue implements IEventQueue { * from being sent after consent is revoked. */ public clear(): void { + // Invalidate any batch already in flight so a later send failure cannot + // unshift it back onto the queue we are emptying here. + this.generation++; this.queue = []; this.payloadHashes.clear(); @@ -434,6 +522,15 @@ export class EventQueue implements IEventQueue { this.timer = null; } + // Detach up front rather than at the end: while the drain loop below is + // awaiting a send, a backgrounding app would otherwise queue a flush that + // outlives cleanup(). The closed check in the handler covers the same + // window; removing the listener means the event never reaches it at all. + if (this.appStateSubscription) { + this.appStateSubscription.remove(); + this.appStateSubscription = null; + } + // A flush may already be in flight with its items spliced out of the // queue, which would make the drain loop below see an empty queue and // return before delivery finished. Wait for it to settle first — on a @@ -479,14 +576,13 @@ export class EventQueue implements IEventQueue { logger.debug(`EventQueue: Cleanup completed, flushed ${initialQueueLength - this.queue.length} events`); } + // The AppState listener was already detached at the top of cleanup, and + // `closed` stops anything arming a timer, so nothing can have been + // scheduled since. This is the last-resort clear for a timer that a flush + // in the drain loop above might have left behind. if (this.timer) { clearTimeout(this.timer); this.timer = null; } - - if (this.appStateSubscription) { - this.appStateSubscription.remove(); - this.appStateSubscription = null; - } } } diff --git a/src/utils/sanitize.ts b/src/utils/sanitize.ts index a222a07..986bdfb 100644 --- a/src/utils/sanitize.ts +++ b/src/utils/sanitize.ts @@ -30,8 +30,9 @@ import type { ITrafficSource } from "../types"; * there it is a browser-set `document.referrer` already handled by redactUrl. * In React Native `referrer` holds the raw deep-link URL the attacker * supplied, so sanitizing only the utm_ and ref fields would still let the - * payload through. It gets the same character rules with a URL-sized budget; - * a well-formed URL percent-encodes the rejected characters anyway. + * payload through. It gets a URL-sized length budget and is checked both raw + * and percent-decoded — a URL encodes its payload, so a raw-only check would + * pass `?utm_source=%3Cscript%3E` straight through. * * Invalid values are dropped to "" — the same representation as "parameter * absent" — rather than repaired, so a poisoned value can never be persisted @@ -57,25 +58,193 @@ const REFERRER_MAX_LENGTH = 2_048; const FORBIDDEN_CHARS = /[<>"'`\\\u0000-\u001f\u007f-\u009f\u200b-\u200f\u2028-\u202e\u2060\ufeff\ufffd]/; +// Applied to the percent-DECODED form of a referrer. Narrower than the raw +// set on purpose: a decoded query value legitimately contains quotes and +// backslashes (`?q=%22running%20shoes%22`), and those are harmless in an +// analytics field. Markup and invisible characters are what indicate an +// injected payload rather than a real deep link. +const DECODED_FORBIDDEN_CHARS = + /[<>\u0000-\u001f\u007f-\u009f\u200b-\u200f\u2028-\u202e\u2060\ufeff\ufffd]/; + // Values smuggling an executable/URL scheme, e.g. `javascript:alert(1)`. const FORBIDDEN_SCHEME_PREFIX = /^\s*(javascript|data|vbscript):/i; +/** + * Whether a value smuggles a dangerous scheme, anywhere a URL parser would + * consider the start of a value. + * + * The pattern is anchored, so testing the string as a whole is not enough once + * decoding can reveal separators that were themselves encoded: with the `=` in + * `?utm_source%3Djavascript%253Aalert(1)` encoded, URLSearchParams reports one + * key that decodes to `utm_source=javascript:alert(1)`, which the anchored test + * never matches. Splitting on the separators after decoding puts the payload + * back at the start of a segment. + */ +const hasForbiddenScheme = (value: string): boolean => + FORBIDDEN_SCHEME_PREFIX.test(value) || + value.split(/[?&=#]/).some((segment) => FORBIDDEN_SCHEME_PREFIX.test(segment)); + +/** + * Percent-decode one layer without ever throwing. `decodeURIComponent` rejects + * the entire string on a single malformed escape (a stray `%`), which would let + * `...%3Cscript%3E%` skip the decoded checks completely. + * + * Each *run* of consecutive escapes is decoded as a unit rather than byte by + * byte, because a multi-byte character spans several escapes: `%E3%80%80` is + * one ideographic space, and decoding `%E3` alone throws. Byte-wise decoding + * would therefore never produce the whitespace that lets `^\s*javascript:` + * match `%E3%80%80javascript%3Aalert(1)`. A run that is not valid UTF-8, and + * any stray `%`, is left as-is. + */ +const decodeRun = (run: string): string => { + try { + return decodeURIComponent(run); + } catch { + // One invalid byte anywhere in the run would otherwise blind the whole + // run — appending `%C0` next to `%3C` is enough to hide a `<`. Decode the + // largest valid group at each position instead, longest first so multi-byte + // sequences (up to four escapes) still group correctly, and pass through + // any escape that cannot be decoded at all. + const escapes = run.match(/%[0-9A-Fa-f]{2}/g) ?? []; + let out = ""; + let i = 0; + while (i < escapes.length) { + let taken = 0; + for (let len = Math.min(4, escapes.length - i); len >= 1; len--) { + try { + out += decodeURIComponent(escapes.slice(i, i + len).join("")); + taken = len; + break; + } catch { + // Try a shorter group. + } + } + if (taken === 0) { + out += escapes[i]; + taken = 1; + } + i += taken; + } + return out; + } +}; + +const decodeOnce = (value: string): string => + value.replace(/(?:%[0-9A-Fa-f]{2})+/g, decodeRun); + +/** + * Decode to a fixed point, so a payload encoded any number of times + * (`%253Cscript%253E`, `%25252525253Cscript...`) is compared in a form the + * markup check can see. + * + * The bound is the input length rather than a fixed number of layers, because + * a fixed number is reachable: re-encoding `javascript:alert(1)` only grows the + * single `%` by two characters per layer, so 66 layers fit in 151 characters + * and any small constant can be encoded past. Every productive pass turns a + * three-character escape into one character, so a string of length n admits + * fewer than n productive passes — this bound is always sufficient and still + * guarantees termination. + */ +const decodeDeep = (value: string): string => { + let current = value; + for (let i = 0; i < value.length; i++) { + const next = decodeOnce(current); + if (next === current) break; + current = next; + } + return current; +}; + const sanitizeRef = (value: string): string => REF_PATTERN.test(value) ? value : ""; -const sanitizeUtm = (value: string): string => - value.length <= UTM_MAX_LENGTH && - !FORBIDDEN_CHARS.test(value) && - !FORBIDDEN_SCHEME_PREFIX.test(value) - ? value - : ""; - -const sanitizeReferrer = (value: string): string => - value.length <= REFERRER_MAX_LENGTH && - !FORBIDDEN_CHARS.test(value) && - !FORBIDDEN_SCHEME_PREFIX.test(value) - ? value - : ""; +/** + * URLSearchParams has already decoded one layer by the time a UTM value gets + * here, so a payload encoded twice arrives still encoded: + * `utm_source=%253Cscript%253E` reads as the literal `%3Cscript%3E`, which + * contains no raw markup and would pass a raw-only check. Test the decoded + * form as well. + */ +const sanitizeUtm = (value: string): string => { + if (value.length > UTM_MAX_LENGTH) return ""; + if (FORBIDDEN_CHARS.test(value) || hasForbiddenScheme(value)) { + return ""; + } + const decoded = decodeDeep(value); + if ( + DECODED_FORBIDDEN_CHARS.test(decoded) || + hasForbiddenScheme(decoded) + ) { + return ""; + } + return value; +}; + +/** + * The decoded parts of a referrer's query string — keys as well as values. + * Handles both a full URL ("myapp://x?utm_source=...") and the bare query + * string the Android install referrer supplies ("utm_source=...&utm_medium="). + * Returns nothing when there is no query to read; the caller still checks the + * whole string. + * + * Keys matter because a parameter with no "=" parses entirely as a key with an + * empty value: "?%6Aavascript:alert(1)" would otherwise contribute nothing to + * inspect. + */ +const referrerParamParts = (value: string): string[] => { + const queryStart = value.indexOf("?"); + const query = queryStart === -1 ? value : value.slice(queryStart + 1); + if (!query || (queryStart === -1 && !query.includes("="))) return []; + try { + // URLSearchParams decodes leniently and does not throw on a stray "%". + const parts: string[] = []; + for (const [key, paramValue] of new URLSearchParams(query)) { + parts.push(key, paramValue); + } + return parts; + } catch { + return []; + } +}; + +/** + * A URL carries its payload percent-encoded — `Linking` and the browser both + * encode `<` and `>` — so a raw-only check would pass + * `?utm_source=%3Cscript%3E` straight through. + * + * Checked three ways, because pattern-matching the URL as one opaque string + * misses what a structural read catches: the whole value raw, the whole value + * decoded, and each decoded query part — keys included — on its own. The last + * is what catches a smuggled scheme, since FORBIDDEN_SCHEME_PREFIX is anchored + * and `?utm_source=javascript:alert(1)` only matches once that parameter is + * read apart from the `myapp://` URL containing it. + */ +const sanitizeReferrer = (value: string): string => { + if (value.length > REFERRER_MAX_LENGTH) return ""; + if (FORBIDDEN_CHARS.test(value) || hasForbiddenScheme(value)) { + return ""; + } + + const decodedWhole = decodeDeep(value); + if ( + DECODED_FORBIDDEN_CHARS.test(decodedWhole) || + hasForbiddenScheme(decodedWhole) + ) { + return ""; + } + + for (const part of referrerParamParts(value)) { + const decoded = decodeDeep(part); + if ( + DECODED_FORBIDDEN_CHARS.test(decoded) || + hasForbiddenScheme(decoded) + ) { + return ""; + } + } + + return value; +}; /** * Sanitize every traffic-source field of a (possibly sparse) traffic-source From 741333dad27eb6ee86759322f87797965b2dae59 Mon Sep 17 00:00:00 2001 From: Yos Riady Date: Mon, 10 Aug 2026 11:42:21 +0700 Subject: [PATCH 06/12] Bound cleanup's wait on an in-flight send MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses Codex review feedback on #83. cleanup() awaits the flush mutex so an in-flight batch — whose items are already spliced out of the queue — settles before the drain loop measures what is left. That wait was unbounded, and React Native's fetch has no request timeout, so a connection that never settles hung teardown forever. The consequence reaches further than a stuck cleanup(): FormoAnalyticsProvider awaits the pending cleanup before constructing a replacement instance (FormoAnalyticsProvider.tsx), so the SDK could never be reconfigured again for the life of the process. The wait is now capped at 5s. Skipping straight to the drain loop on timeout would not have helped — flush() awaits the same stalled mutex and would hang identically — so a timed-out wait abandons the queued events and finishes teardown instead, which matches how the existing safety-limit path gives up. The regression test stalls fetch with a promise that never settles; it hits Jest's timeout with the bound removed and completes with it in place. 364 tests pass (363 before), typecheck, lint and `pnpm audit --prod` clean. Co-Authored-By: Claude Opus 5 --- src/__tests__/eventQueue.test.ts | 39 ++++++++++++++++++++++++++ src/lib/event/EventQueue.ts | 48 ++++++++++++++++++++++++++++++-- 2 files changed, 85 insertions(+), 2 deletions(-) diff --git a/src/__tests__/eventQueue.test.ts b/src/__tests__/eventQueue.test.ts index 2749b4a..8098d05 100644 --- a/src/__tests__/eventQueue.test.ts +++ b/src/__tests__/eventQueue.test.ts @@ -543,6 +543,45 @@ describe("EventQueue", () => { }); }); + describe("cleanup with a stalled request", () => { + it("completes rather than hanging when the send never settles", async () => { + jest.useFakeTimers(); + try { + // A connection that never resolves and never rejects. React Native's + // fetch has no request timeout, so nothing else will end this. + fetchMock.mockReturnValue(new Promise(() => {})); + + const queue = new EventQueue("test-write-key", { + apiHost: "https://events.formo.test", + flushAt: 20, + flushInterval: 10_000, + retryCount: 1, + }); + + // The first event flushes immediately and hangs mid-send. + await queue.enqueue(makeEvent(1)); + await jest.advanceTimersByTimeAsync(100); + expect(fetchMock).toHaveBeenCalledTimes(1); + + let done = false; + const cleanup = queue.cleanup().then(() => { + done = true; + }); + + // Still stuck on the in-flight send. + await jest.advanceTimersByTimeAsync(1_000); + expect(done).toBe(false); + + // Past the bound, teardown gives up on it and finishes. + await jest.advanceTimersByTimeAsync(10_000); + await cleanup; + expect(done).toBe(true); + } finally { + jest.useRealTimers(); + } + }); + }); + describe("enqueue racing cleanup", () => { it("drops an event whose hashing was still pending when cleanup ran", async () => { const queue = makeQueue(); diff --git a/src/lib/event/EventQueue.ts b/src/lib/event/EventQueue.ts index a3d591d..f79ad04 100644 --- a/src/lib/event/EventQueue.ts +++ b/src/lib/event/EventQueue.ts @@ -51,6 +51,11 @@ const DEFAULT_QUEUE_SIZE = 1_024 * 500; // 500kB const MAX_QUEUE_SIZE = 1_024 * 500; // 500kB const MIN_QUEUE_SIZE = 200; // 200 bytes +// How long cleanup() waits for an already-in-flight send before abandoning it. +// Teardown must finish promptly: the provider blocks re-initialization on the +// pending cleanup, so an unbounded wait would strand the SDK. +const CLEANUP_FLUSH_WAIT = 1_000 * 5; // 5 seconds + const DEFAULT_FLUSH_INTERVAL = 1_000 * 30; // 30 seconds const MAX_FLUSH_INTERVAL = 1_000 * 300; // 5 minutes const MIN_FLUSH_INTERVAL = 1_000 * 10; // 10 seconds @@ -511,6 +516,25 @@ export class EventQueue implements IEventQueue { logger.debug("EventQueue: Cleared all pending events"); } + /** + * Wait for an in-flight flush to settle, giving up after CLEANUP_FLUSH_WAIT. + * Returns whether it settled. The timer is always cleared, so a prompt + * settle does not leave one pending. + */ + private async awaitInFlightFlush(): Promise { + let timer: ReturnType | undefined; + try { + return await Promise.race([ + this.flushMutex.then(() => true), + new Promise((resolve) => { + timer = setTimeout(() => resolve(false), CLEANUP_FLUSH_WAIT); + }), + ]); + } finally { + if (timer) clearTimeout(timer); + } + } + /** * Clean up resources, flushing any pending events first */ @@ -535,8 +559,28 @@ export class EventQueue implements IEventQueue { // queue, which would make the drain loop below see an empty queue and // return before delivery finished. Wait for it to settle first — on a // transient failure it puts those items back, and the loop then retries - // them. The mutex always resolves, including on a failed send. - await this.flushMutex; + // them. + // + // Bounded, because the mutex only resolves once the send settles and + // `fetch` here has no request timeout: a stalled connection would + // otherwise hang cleanup() forever, and FormoAnalyticsProvider awaits the + // pending cleanup before building a replacement instance, so the SDK could + // never be reconfigured again. + const settled = await this.awaitInFlightFlush(); + + if (!settled) { + // The drain loop cannot help here: flush() awaits the same stalled + // mutex, so it would hang exactly as this wait just did. Give up the + // queued events rather than the teardown. + logger.warn( + `EventQueue: In-flight flush did not settle within ${millisecondsToSecond( + CLEANUP_FLUSH_WAIT + )}s, abandoning ${this.queue.length} event(s)` + ); + this.queue = []; + this.payloadHashes.clear(); + return; + } // Flush all remaining queued events before teardown // Loop until queue is empty since flush() only sends flushAt events per call From 124347b30f1022f702d317f1d6e0edcb9bc570a6 Mon Sep 17 00:00:00 2001 From: Yos Riady Date: Mon, 10 Aug 2026 11:51:30 +0700 Subject: [PATCH 07/12] Invalidate the abandoned batch when cleanup times out MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to 741333d, which bounded cleanup()'s wait but emptied the queue by hand instead of going through clear(). Only clear() bumps the generation, and that bump is what actually invalidates a batch still held by a stalled flush. Without it: flush A splices event 1 and stalls mid-send. flush B queues behind it on the mutex. cleanup() waits, times out after 5s, empties the queue and returns. A finally fails retryably, sees an unchanged generation, and unshifts event 1 back onto the queue teardown just emptied. B then wakes, finds it, sends it, and invokes its callback a second time — after cleanup() had already resolved and the provider had moved on to a replacement instance. Both discard paths in cleanup() now call clear(), so the stalled flush takes its discard branch and the chained flush finds nothing to send. The regression test stalls the first send, queues a second flush behind it, lets cleanup time out, and only then fails the stalled request: the event's callback fires twice without this fix and once with it. 365 tests pass (364 before), typecheck, lint and `pnpm audit --prod` clean. Co-Authored-By: Claude Opus 5 --- src/__tests__/eventQueue.test.ts | 48 ++++++++++++++++++++++++++++++++ src/lib/event/EventQueue.ts | 15 +++++++--- 2 files changed, 59 insertions(+), 4 deletions(-) diff --git a/src/__tests__/eventQueue.test.ts b/src/__tests__/eventQueue.test.ts index 8098d05..527a6e5 100644 --- a/src/__tests__/eventQueue.test.ts +++ b/src/__tests__/eventQueue.test.ts @@ -582,6 +582,54 @@ describe("EventQueue", () => { }); }); + describe("cleanup abandoning a stalled batch", () => { + it("does not let a stalled flush resurrect and resend its events", async () => { + jest.useFakeTimers(); + try { + let releaseStall: (value: { ok: boolean; status: number }) => void; + fetchMock.mockReturnValueOnce( + new Promise((resolve) => { + releaseStall = resolve; + }) + ); + // Everything after the stall fails retryably, so the stalled flush + // exhausts its retries and takes the re-queue path. + fetchMock.mockResolvedValue({ ok: false, status: 500 }); + + const queue = new EventQueue("test-write-key", { + apiHost: "https://events.formo.test", + flushAt: 1, + flushInterval: 10_000, + retryCount: 1, + }); + + // Flush A splices event 1 and stalls mid-send. + const callback = jest.fn(); + await queue.enqueue(makeEvent(1), callback); + await jest.advanceTimersByTimeAsync(100); + expect(fetchMock).toHaveBeenCalledTimes(1); + + // Flush B is queued behind A, waiting on its mutex. + await queue.enqueue(makeEvent(2)); + + // Teardown gives up on A after the bound and returns. + const cleanup = queue.cleanup(); + await jest.advanceTimersByTimeAsync(10_000); + await cleanup; + + // A only now fails. It must not put event 1 back for B to send. + releaseStall!({ ok: false, status: 500 }); + await jest.advanceTimersByTimeAsync(60_000); + + // A reports the failure once. A second invocation means B resurrected + // the event and sent it after teardown had already completed. + expect(callback).toHaveBeenCalledTimes(1); + } finally { + jest.useRealTimers(); + } + }); + }); + describe("enqueue racing cleanup", () => { it("drops an event whose hashing was still pending when cleanup ran", async () => { const queue = makeQueue(); diff --git a/src/lib/event/EventQueue.ts b/src/lib/event/EventQueue.ts index f79ad04..eff0d04 100644 --- a/src/lib/event/EventQueue.ts +++ b/src/lib/event/EventQueue.ts @@ -577,8 +577,14 @@ export class EventQueue implements IEventQueue { CLEANUP_FLUSH_WAIT )}s, abandoning ${this.queue.length} event(s)` ); - this.queue = []; - this.payloadHashes.clear(); + // clear() rather than emptying the queue by hand: it also bumps the + // generation, which is what actually invalidates the stalled batch. That + // batch still holds its items, and a flush chained behind it is still + // waiting on its mutex. Without the bump, the stalled flush would later + // fail, unshift its items back onto the queue we just emptied, and the + // chained flush would then send them and invoke their callbacks a second + // time — after teardown had already returned. + this.clear(); return; } @@ -612,8 +618,9 @@ export class EventQueue implements IEventQueue { logger.warn( `EventQueue: Cleanup safety limit reached. Discarding ${this.queue.length} events.` ); - this.queue = []; - this.payloadHashes.clear(); + // Same reasoning as the timeout path above: bump the generation so a + // flush still holding these items cannot put them back after teardown. + this.clear(); } if (initialQueueLength > 0) { From 3b352b50c2afb8fb6444efecce4e32e8c546517e Mon Sep 17 00:00:00 2001 From: Yos Riady Date: Mon, 10 Aug 2026 12:04:12 +0700 Subject: [PATCH 08/12] Bound the whole of cleanup, not just the in-flight flush MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses Codex review feedback on #83. 741333d bounded the wait on a flush that was already in flight, but that is the less common case and left the ordinary one unbounded: with nothing in flight the wait returns immediately, and the drain loop then opens a fresh request of its own. React Native's fetch has no request timeout, so a stall there hung teardown exactly as before — and with it the provider, which awaits the pending cleanup before constructing a replacement instance. cleanup() now runs against a single deadline covering every send it waits on, rather than a bound per wait. That also keeps total teardown bounded instead of scaling at 5s per flush attempt. The drain loop's flush is folded into that race as a settled promise rather than a rejecting one, so only the deadline can win on error and no rejection escapes unhandled. Its error branch keeps the previous behaviour of logging and breaking out of the loop. Both give-up paths — deadline exceeded and the pre-existing safety limit — now share abandonQueuedEvents(), which routes through clear() so a send still holding a batch cannot put it back after teardown returns. The regression test lets the first send succeed so nothing is in flight, leaving the stall to the request the drain loop itself opens: it hits Jest's timeout with the bound removed and completes with it in place. 366 tests pass (365 before), typecheck, lint and `pnpm audit --prod` clean. Co-Authored-By: Claude Opus 5 --- src/__tests__/eventQueue.test.ts | 44 ++++++++ src/lib/event/EventQueue.ts | 181 ++++++++++++++++--------------- 2 files changed, 140 insertions(+), 85 deletions(-) diff --git a/src/__tests__/eventQueue.test.ts b/src/__tests__/eventQueue.test.ts index 527a6e5..da6018e 100644 --- a/src/__tests__/eventQueue.test.ts +++ b/src/__tests__/eventQueue.test.ts @@ -630,6 +630,50 @@ describe("EventQueue", () => { }); }); + describe("cleanup with a stalled drain flush", () => { + it("completes when a flush cleanup itself starts never settles", async () => { + jest.useFakeTimers(); + try { + // First send succeeds, so nothing is in flight when cleanup begins and + // the in-flight wait returns immediately. The stall happens on the + // request the drain loop opens. + fetchMock.mockResolvedValueOnce({ ok: true, status: 200 }); + fetchMock.mockReturnValue(new Promise(() => {})); + + const queue = new EventQueue("test-write-key", { + apiHost: "https://events.formo.test", + flushAt: 20, + flushInterval: 10_000, + retryCount: 1, + }); + + await queue.enqueue(makeEvent(1)); + await jest.advanceTimersByTimeAsync(100); + expect(fetchMock).toHaveBeenCalledTimes(1); + + // Queue a second event, but do not let its interval fire — it is still + // sitting in the queue with no request in flight. + await queue.enqueue(makeEvent(2)); + + let done = false; + const cleanup = queue.cleanup().then(() => { + done = true; + }); + + // The drain loop opens a request for event 2, which never settles. + await jest.advanceTimersByTimeAsync(1_000); + expect(done).toBe(false); + + // Teardown must still finish once the deadline passes. + await jest.advanceTimersByTimeAsync(10_000); + await cleanup; + expect(done).toBe(true); + } finally { + jest.useRealTimers(); + } + }); + }); + describe("enqueue racing cleanup", () => { it("drops an event whose hashing was still pending when cleanup ran", async () => { const queue = makeQueue(); diff --git a/src/lib/event/EventQueue.ts b/src/lib/event/EventQueue.ts index eff0d04..14e4f33 100644 --- a/src/lib/event/EventQueue.ts +++ b/src/lib/event/EventQueue.ts @@ -517,22 +517,20 @@ export class EventQueue implements IEventQueue { } /** - * Wait for an in-flight flush to settle, giving up after CLEANUP_FLUSH_WAIT. - * Returns whether it settled. The timer is always cleared, so a prompt - * settle does not leave one pending. + * Give up on the events still queued at teardown, and make sure a send that + * is still holding a batch cannot put them back. */ - private async awaitInFlightFlush(): Promise { - let timer: ReturnType | undefined; - try { - return await Promise.race([ - this.flushMutex.then(() => true), - new Promise((resolve) => { - timer = setTimeout(() => resolve(false), CLEANUP_FLUSH_WAIT); - }), - ]); - } finally { - if (timer) clearTimeout(timer); - } + private abandonQueuedEvents(reason: string): void { + logger.warn( + `EventQueue: ${reason}, abandoning ${this.queue.length} event(s)` + ); + // clear() rather than emptying the queue by hand: it also bumps the + // generation, which is what actually invalidates a batch a stalled flush + // is still holding. Without the bump that flush would later fail, unshift + // its items back onto the queue we just emptied, and a flush chained + // behind it would send them and invoke their callbacks a second time — + // after teardown had already returned. + this.clear(); } /** @@ -555,85 +553,98 @@ export class EventQueue implements IEventQueue { this.appStateSubscription = null; } - // A flush may already be in flight with its items spliced out of the - // queue, which would make the drain loop below see an empty queue and - // return before delivery finished. Wait for it to settle first — on a - // transient failure it puts those items back, and the loop then retries - // them. + // One deadline for the whole of teardown, not per wait. // - // Bounded, because the mutex only resolves once the send settles and - // `fetch` here has no request timeout: a stalled connection would - // otherwise hang cleanup() forever, and FormoAnalyticsProvider awaits the - // pending cleanup before building a replacement instance, so the SDK could - // never be reconfigured again. - const settled = await this.awaitInFlightFlush(); - - if (!settled) { - // The drain loop cannot help here: flush() awaits the same stalled - // mutex, so it would hang exactly as this wait just did. Give up the - // queued events rather than the teardown. - logger.warn( - `EventQueue: In-flight flush did not settle within ${millisecondsToSecond( - CLEANUP_FLUSH_WAIT - )}s, abandoning ${this.queue.length} event(s)` - ); - // clear() rather than emptying the queue by hand: it also bumps the - // generation, which is what actually invalidates the stalled batch. That - // batch still holds its items, and a flush chained behind it is still - // waiting on its mutex. Without the bump, the stalled flush would later - // fail, unshift its items back onto the queue we just emptied, and the - // chained flush would then send them and invoke their callbacks a second - // time — after teardown had already returned. - this.clear(); - return; - } + // `fetch` here has no request timeout, so any send this method waits on — + // a flush already in flight, or one the drain loop starts itself — can + // stall forever. Bounding only the first would leave the ordinary case + // unbounded: with nothing in flight the first wait returns immediately and + // the drain loop then opens a fresh request. A single deadline also keeps + // total teardown bounded rather than 5s per flush attempt. + // + // This matters beyond a stuck cleanup(): FormoAnalyticsProvider awaits the + // pending cleanup before constructing a replacement instance, so a hang + // here means the SDK can never be reconfigured again. + let deadlineTimer: ReturnType | undefined; + const deadline = new Promise<"timeout">((resolve) => { + deadlineTimer = setTimeout(() => resolve("timeout"), CLEANUP_FLUSH_WAIT); + }); + const withinDeadline = (work: Promise) => + Promise.race([work, deadline]); + const timedOut = `Teardown exceeded ${millisecondsToSecond( + CLEANUP_FLUSH_WAIT + )}s`; - // Flush all remaining queued events before teardown - // Loop until queue is empty since flush() only sends flushAt events per call - // Safety limit prevents infinite loops if flush silently fails - const maxAttempts = Math.ceil(this.queue.length / this.flushAt) + 3; - let attempts = 0; - const initialQueueLength = this.queue.length; + try { + // A flush may already be in flight with its items spliced out of the + // queue, which would make the drain loop below see an empty queue and + // return before delivery finished. Wait for it to settle first — on a + // transient failure it puts those items back, and the loop then retries + // them. + if ( + (await withinDeadline(this.flushMutex.then(() => "settled" as const))) === + "timeout" + ) { + this.abandonQueuedEvents(timedOut); + return; + } - while (this.queue.length > 0 && attempts < maxAttempts) { - const queueLengthBefore = this.queue.length; - try { - await this.flush(); - } catch (error) { - logger.error("EventQueue: Failed to flush during cleanup", error); + // Flush all remaining queued events before teardown + // Loop until queue is empty since flush() only sends flushAt events per call + // Safety limit prevents infinite loops if flush silently fails + const maxAttempts = Math.ceil(this.queue.length / this.flushAt) + 3; + let attempts = 0; + const initialQueueLength = this.queue.length; + + while (this.queue.length > 0 && attempts < maxAttempts) { + const queueLengthBefore = this.queue.length; + + const outcome = await withinDeadline( + // Settle rather than reject, so only the deadline can win the race + // on an error and a rejection cannot escape unhandled. + this.flush().then( + () => "flushed" as const, + (error) => { + logger.error("EventQueue: Failed to flush during cleanup", error); + return "failed" as const; + } + ) + ); + + if (outcome === "timeout") { + this.abandonQueuedEvents(timedOut); + return; + } // Break on error to avoid infinite loop if flush keeps failing - break; - } + if (outcome === "failed") break; - // If queue length didn't decrease, flush is silently failing - if (this.queue.length >= queueLengthBefore) { - logger.warn("EventQueue: Flush did not reduce queue size, aborting cleanup"); - break; - } + // If queue length didn't decrease, flush is silently failing + if (this.queue.length >= queueLengthBefore) { + logger.warn("EventQueue: Flush did not reduce queue size, aborting cleanup"); + break; + } - attempts++; - } + attempts++; + } - if (attempts >= maxAttempts && this.queue.length > 0) { - logger.warn( - `EventQueue: Cleanup safety limit reached. Discarding ${this.queue.length} events.` - ); - // Same reasoning as the timeout path above: bump the generation so a - // flush still holding these items cannot put them back after teardown. - this.clear(); - } + if (attempts >= maxAttempts && this.queue.length > 0) { + this.abandonQueuedEvents("Cleanup safety limit reached"); + } - if (initialQueueLength > 0) { - logger.debug(`EventQueue: Cleanup completed, flushed ${initialQueueLength - this.queue.length} events`); - } + if (initialQueueLength > 0) { + logger.debug(`EventQueue: Cleanup completed, flushed ${initialQueueLength - this.queue.length} events`); + } + } finally { + if (deadlineTimer) clearTimeout(deadlineTimer); - // The AppState listener was already detached at the top of cleanup, and - // `closed` stops anything arming a timer, so nothing can have been - // scheduled since. This is the last-resort clear for a timer that a flush - // in the drain loop above might have left behind. - if (this.timer) { - clearTimeout(this.timer); - this.timer = null; + // The AppState listener was already detached at the top of cleanup, and + // `closed` stops anything arming a timer, so nothing can have been + // scheduled since. This is the last-resort clear for a timer that a flush + // in the drain loop above might have left behind. + if (this.timer) { + clearTimeout(this.timer); + this.timer = null; + } } } } From b80f53c1f06c6a057321e7d9e8cd167dfc9eb406 Mon Sep 17 00:00:00 2001 From: Yos Riady Date: Mon, 10 Aug 2026 12:12:00 +0700 Subject: [PATCH 09/12] Suppress callbacks for a batch cleanup abandoned MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to 3b352b5. Bumping the generation stopped an abandoned batch being re-queued, but not its callbacks: the request cleanup() gave up on has no timeout, so it can settle long after teardown resolved, and flush() then still ran done() on it. The consumer's per-event callback fired against an instance the app had already torn down, and which the provider may since have replaced. done() now returns early when the queue is closed and the batch's generation is stale — that pair means specifically "abandoned during teardown". A plain clear() for opt-out is unaffected, so a consumer still learns those events were dropped. This also tightens the earlier resurrection test: after teardown the correct expectation is no callback at all rather than exactly one, and it still fails if the abandoned batch is resurrected. 367 tests pass (366 before), typecheck, lint and `pnpm audit --prod` clean. Co-Authored-By: Claude Opus 5 --- src/__tests__/eventQueue.test.ts | 49 ++++++++++++++++++++++++++++++-- src/lib/event/EventQueue.ts | 7 +++++ 2 files changed, 54 insertions(+), 2 deletions(-) diff --git a/src/__tests__/eventQueue.test.ts b/src/__tests__/eventQueue.test.ts index da6018e..e6caaee 100644 --- a/src/__tests__/eventQueue.test.ts +++ b/src/__tests__/eventQueue.test.ts @@ -621,9 +621,10 @@ describe("EventQueue", () => { releaseStall!({ ok: false, status: 500 }); await jest.advanceTimersByTimeAsync(60_000); - // A reports the failure once. A second invocation means B resurrected + // Nothing may call back into a torn-down instance: A's batch was + // abandoned by cleanup, and any invocation here means B resurrected // the event and sent it after teardown had already completed. - expect(callback).toHaveBeenCalledTimes(1); + expect(callback).not.toHaveBeenCalled(); } finally { jest.useRealTimers(); } @@ -674,6 +675,50 @@ describe("EventQueue", () => { }); }); + describe("callbacks after an abandoned teardown", () => { + it("does not invoke callbacks when the abandoned request finally settles", async () => { + jest.useFakeTimers(); + try { + let releaseStall: (value: { ok: boolean; status: number }) => void; + fetchMock.mockResolvedValueOnce({ ok: true, status: 200 }); + fetchMock.mockReturnValue( + new Promise((resolve) => { + releaseStall = resolve; + }) + ); + + const queue = new EventQueue("test-write-key", { + apiHost: "https://events.formo.test", + flushAt: 20, + flushInterval: 10_000, + retryCount: 1, + }); + + await queue.enqueue(makeEvent(1)); + await jest.advanceTimersByTimeAsync(100); + + // Queued with a callback, no request in flight. + const callback = jest.fn(); + await queue.enqueue(makeEvent(2), callback); + + // The drain loop opens a request for it, which stalls past the + // deadline; teardown abandons it and returns. + const cleanup = queue.cleanup(); + await jest.advanceTimersByTimeAsync(10_000); + await cleanup; + expect(callback).not.toHaveBeenCalled(); + + // The abandoned request settles long afterwards. The app has already + // torn the SDK down, so nothing may call back into it. + releaseStall!({ ok: true, status: 200 }); + await jest.advanceTimersByTimeAsync(60_000); + expect(callback).not.toHaveBeenCalled(); + } finally { + jest.useRealTimers(); + } + }); + }); + describe("enqueue racing cleanup", () => { it("drops an event whose hashing was still pending when cleanup ran", async () => { const queue = makeQueue(); diff --git a/src/lib/event/EventQueue.ts b/src/lib/event/EventQueue.ts index 14e4f33..550dd66 100644 --- a/src/lib/event/EventQueue.ts +++ b/src/lib/event/EventQueue.ts @@ -358,6 +358,13 @@ export class EventQueue implements IEventQueue { })); const done = (err?: Error) => { + // A batch cleanup() abandoned must not call back into an app that has + // already torn the SDK down: its request has no timeout, so it can + // settle long after cleanup() resolved and the provider moved on to a + // replacement instance. A plain clear() (opt-out) still notifies, so + // the consumer does learn those events were dropped. + if (this.closed && this.generation !== generation) return; + items.forEach(({ message, callback: itemCallback }) => safeCall(itemCallback, err, message, data) ); From caecaba9ccc0e9fbf96c88bcfe2423396d9dab6f Mon Sep 17 00:00:00 2001 From: Yos Riady Date: Mon, 10 Aug 2026 12:32:52 +0700 Subject: [PATCH 10/12] Abandon flushes that outlive cleanup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Second Codex finding on 3b352b5: a flush waiting on the mutex when teardown begins could still send. Scenario: cleanup()'s drain flush is in flight; a consumer flush() (or a second cleanup) queues behind it on the mutex; the drain flush fails retryably and re-queues its events; cleanup breaks out of its loop and resolves. The queued flush then takes the mutex and sends those events, and invokes their callbacks, after teardown had finished. Two changes: Every give-up path in cleanup now empties and invalidates the queue, not just the safety-limit one, so nothing is left behind for a later flush to find. On its own this was not sufficient, which the first version of the regression test failed to show: the queued flush takes the mutex the instant the drain flush releases it — before cleanup has even observed that failure — so the events are already gone by the time cleanup clears anything. So flush() now abandons itself if teardown began while it was waiting its turn. That check has to identify cleanup's own drain calls specifically rather than a window of time: cleanup spends most of its drain loop awaiting one of them, and a `draining` flag covering that window classified the racing flush as internal. The flush body moved to a private runFlush(), which cleanup calls with duringCleanup: true; the public flush() is unchanged for callers. 368 tests pass (367 before), typecheck, lint and `pnpm audit --prod` clean. The regression test asserts absolute send counts rather than a delta measured after cleanup() resolves, because the racing send lands before that: 3 with the guard, 5 without. Co-Authored-By: Claude Opus 5 --- src/__tests__/eventQueue.test.ts | 67 ++++++++++++++++++++++++++++++++ src/lib/event/EventQueue.ts | 45 ++++++++++++++++++--- 2 files changed, 107 insertions(+), 5 deletions(-) diff --git a/src/__tests__/eventQueue.test.ts b/src/__tests__/eventQueue.test.ts index e6caaee..4d2a3d8 100644 --- a/src/__tests__/eventQueue.test.ts +++ b/src/__tests__/eventQueue.test.ts @@ -719,6 +719,73 @@ describe("EventQueue", () => { }); }); + describe("flush racing cleanup", () => { + it("does not let a flush queued behind cleanup send abandoned events", async () => { + jest.useFakeTimers(); + try { + let releaseDrain: (value: { ok: boolean; status: number }) => void; + // 1st: the first-event flush, succeeds and empties the queue. + fetchMock.mockResolvedValueOnce({ ok: true, status: 200 }); + // 2nd: the request cleanup's drain loop opens — held so a consumer + // flush can queue behind it. + fetchMock.mockReturnValueOnce( + new Promise((resolve) => { + releaseDrain = resolve; + }) + ); + // Everything after fails retryably, so the drain flush re-queues its + // event and cleanup breaks out of the loop instead of emptying it. + fetchMock.mockResolvedValue({ ok: false, status: 500 }); + + const queue = new EventQueue("test-write-key", { + apiHost: "https://events.formo.test", + flushAt: 20, + flushInterval: 10_000, + retryCount: 1, + }); + + await queue.enqueue(makeEvent(1)); + await jest.advanceTimersByTimeAsync(100); + + const callback = jest.fn(); + await queue.enqueue(makeEvent(2), callback); + + // Nothing is in flight, so cleanup's wait returns at once and its + // drain loop opens the held request. + const cleanupPromise = queue.cleanup(); + await jest.advanceTimersByTimeAsync(10); + expect(fetchMock).toHaveBeenCalledTimes(2); + + // A consumer flush now queues behind cleanup's own. + const racing = queue.flush().catch(() => {}); + + // Let the drain flush fail within the deadline and re-queue. + releaseDrain!({ ok: false, status: 500 }); + await jest.advanceTimersByTimeAsync(4_000); + await cleanupPromise; + + // Sends and callbacks up to here were on a live instance. Nothing more + // may happen now that teardown has resolved. + const sendsAtCleanup = fetchMock.mock.calls.length; + const callbacksAtCleanup = callback.mock.calls.length; + + await jest.advanceTimersByTimeAsync(30_000); + await racing; + + // The racing flush takes the mutex the instant cleanup's drain flush + // releases it — before cleanup() has even observed that failure — so + // measuring only after cleanup() resolves would miss its send. Assert + // the absolute count instead: 1 first-event send, then the drain + // flush's attempt and its one retry. The racing flush must add none. + expect(fetchMock).toHaveBeenCalledTimes(3); + expect(sendsAtCleanup).toBe(3); + expect(callbacksAtCleanup).toBe(callback.mock.calls.length); + } finally { + jest.useRealTimers(); + } + }); + }); + describe("enqueue racing cleanup", () => { it("drops an event whose hashing was still pending when cleanup ran", async () => { const queue = makeQueue(); diff --git a/src/lib/event/EventQueue.ts b/src/lib/event/EventQueue.ts index 550dd66..c8c98b6 100644 --- a/src/lib/event/EventQueue.ts +++ b/src/lib/event/EventQueue.ts @@ -323,6 +323,21 @@ export class EventQueue implements IEventQueue { * preventing race conditions with re-queued items on failure. */ async flush(callback?: (...args: unknown[]) => void): Promise { + return this.runFlush(callback, false); + } + + /** + * The flush body. `duringCleanup` marks the calls teardown makes itself, so + * they are exempt from the closed check below — every other flush must be + * abandoned once teardown has begun. It has to identify the specific call + * rather than a window of time, because cleanup spends most of its drain + * loop awaiting one of these, and a concurrent flush arriving then would + * look internal. + */ + private async runFlush( + callback: ((...args: unknown[]) => void) | undefined, + duringCleanup: boolean + ): Promise { callback = callback || noop; if (this.timer) { @@ -341,6 +356,16 @@ export class EventQueue implements IEventQueue { // Wait for any previous flush to complete await previousMutex; + // Teardown began while this flush was waiting its turn. The instant the + // flush ahead released the mutex it would otherwise splice and send — + // before cleanup() had even observed that flush finishing — delivering + // events after teardown resolved. + if (this.closed && !duringCleanup) { + logger.debug("EventQueue: Abandoning flush that outlived cleanup"); + safeCall(callback); + return; + } + if (!this.queue.length) { safeCall(callback); return; @@ -609,7 +634,7 @@ export class EventQueue implements IEventQueue { const outcome = await withinDeadline( // Settle rather than reject, so only the deadline can win the race // on an error and a rejection cannot escape unhandled. - this.flush().then( + this.runFlush(undefined, true).then( () => "flushed" as const, (error) => { logger.error("EventQueue: Failed to flush during cleanup", error); @@ -634,13 +659,23 @@ export class EventQueue implements IEventQueue { attempts++; } - if (attempts >= maxAttempts && this.queue.length > 0) { - this.abandonQueuedEvents("Cleanup safety limit reached"); - } - if (initialQueueLength > 0) { logger.debug(`EventQueue: Cleanup completed, flushed ${initialQueueLength - this.queue.length} events`); } + + // Teardown always ends with an empty, invalidated queue — not only when + // the safety limit is hit. Each `break` above (a failed flush, a flush + // that did not shrink the queue) otherwise left events behind with the + // generation unchanged, and a flush queued behind cleanup's own — a + // concurrent public flush(), or a second cleanup() — would then send + // them and invoke their callbacks after this call had resolved. + if (this.queue.length > 0) { + this.abandonQueuedEvents( + attempts >= maxAttempts + ? "Cleanup safety limit reached" + : "Teardown finished with events still queued" + ); + } } finally { if (deadlineTimer) clearTimeout(deadlineTimer); From 7a282d6b0ef3f19e5d2828fd7107d83bfbe78916 Mon Sep 17 00:00:00 2001 From: Yos Riady Date: Mon, 10 Aug 2026 12:38:55 +0700 Subject: [PATCH 11/12] Make cleanup idempotent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Last finding on caecaba. cleanup()'s own drain flushes are exempt from the "abandon flushes that outlive teardown" check, which is correct for one teardown but not for two: a second cleanup() waiting on the mutex resumed before the first observed its failed drain, spliced the events that drain had just re-queued, and sent them after the first cleanup() resolved. The first run's final abandon could not help — by then the queue was empty because the second run had taken the events. cleanup() now returns the run already under way instead of starting a competing one. Teardown should be idempotent regardless, and cleanup() is public API as well as being called by the provider. 369 tests pass (368 before); the regression test issues two concurrent teardowns and expects three sends, which is five without this change. Co-Authored-By: Claude Opus 5 --- src/__tests__/eventQueue.test.ts | 45 ++++++++++++++++++++++++++++++++ src/lib/event/EventQueue.ts | 16 ++++++++++++ 2 files changed, 61 insertions(+) diff --git a/src/__tests__/eventQueue.test.ts b/src/__tests__/eventQueue.test.ts index 4d2a3d8..7621c51 100644 --- a/src/__tests__/eventQueue.test.ts +++ b/src/__tests__/eventQueue.test.ts @@ -786,6 +786,51 @@ describe("EventQueue", () => { }); }); + describe("concurrent cleanup", () => { + it("joins the teardown already under way instead of racing it", async () => { + jest.useFakeTimers(); + try { + fetchMock.mockResolvedValueOnce({ ok: true, status: 200 }); + // The drain flush fails retryably and re-queues, which is what a second + // teardown would otherwise pick up and send. + fetchMock.mockResolvedValue({ ok: false, status: 500 }); + + const queue = new EventQueue("test-write-key", { + apiHost: "https://events.formo.test", + flushAt: 20, + flushInterval: 10_000, + retryCount: 1, + }); + + await queue.enqueue(makeEvent(1)); + await jest.advanceTimersByTimeAsync(100); + + const callback = jest.fn(); + await queue.enqueue(makeEvent(2), callback); + + // Two teardowns started concurrently — the provider can call cleanup + // more than once, and it is public API besides. + const first = queue.cleanup(); + const second = queue.cleanup(); + await jest.advanceTimersByTimeAsync(30_000); + await Promise.all([first, second]); + + const sendsAfterTeardown = fetchMock.mock.calls.length; + const callbacksAfterTeardown = callback.mock.calls.length; + + await jest.advanceTimersByTimeAsync(30_000); + expect(fetchMock.mock.calls.length).toBe(sendsAfterTeardown); + expect(callback.mock.calls.length).toBe(callbacksAfterTeardown); + + // 1 first-event send, then the single drain flush and its one retry. + // A second teardown would add its own attempts on top. + expect(fetchMock).toHaveBeenCalledTimes(3); + } finally { + jest.useRealTimers(); + } + }); + }); + describe("enqueue racing cleanup", () => { it("drops an event whose hashing was still pending when cleanup ran", async () => { const queue = makeQueue(); diff --git a/src/lib/event/EventQueue.ts b/src/lib/event/EventQueue.ts index c8c98b6..bbe1a9e 100644 --- a/src/lib/event/EventQueue.ts +++ b/src/lib/event/EventQueue.ts @@ -120,6 +120,13 @@ export class EventQueue implements IEventQueue { * after consent was withdrawn. */ private generation = 0; + /** + * The in-progress teardown, if any. cleanup() is public and the provider can + * call it more than once; a second run would start its own drain flush that + * is exempt from the closed check, splice the events the first run had just + * re-queued, and send them after the first cleanup() had already resolved. + */ + private cleanupPromise: Promise | null = null; constructor(writeKey: string, options: Options) { this.writeKey = writeKey; @@ -569,6 +576,15 @@ export class EventQueue implements IEventQueue { * Clean up resources, flushing any pending events first */ public async cleanup(): Promise { + // Teardown is idempotent: a caller that asks twice joins the run already + // under way rather than starting a competing one. + if (!this.cleanupPromise) { + this.cleanupPromise = this.runCleanup(); + } + return this.cleanupPromise; + } + + private async runCleanup(): Promise { // Stop anything from arming a new timer for the rest of teardown. this.closed = true; if (this.timer) { From 79699dacd73a4effe80f197db52067144430637b Mon Sep 17 00:00:00 2001 From: Yos Riady Date: Mon, 10 Aug 2026 12:46:23 +0700 Subject: [PATCH 12/12] Harden teardown against a throwing listener and a late callback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two follow-ups on 7a282d6. A rejection from runCleanup() was memoised, so every later cleanup() would return that same rejected promise and never retry — and the provider, which awaits the pending cleanup before constructing a replacement, would reject with it forever. Teardown is best-effort and now always settles, clearing the queue on the failure path so nothing is left for a straggling flush. The AppState listener removal is also guarded, since that native call is the one thing that can throw before the queue is drained or invalidated. The abandon path in runFlush() no longer invokes the caller's callback. It resumes only once the flush ahead of it settles, which can be long after cleanup() resolved, and nothing may call back into a torn-down instance. The returned promise still resolves, so an awaiting caller is not left hanging. 371 tests pass (369 before), typecheck, lint and `pnpm audit --prod` clean. Co-Authored-By: Claude Opus 5 --- src/__tests__/eventQueue.test.ts | 54 ++++++++++++++++++++++++++++++++ src/lib/event/EventQueue.ts | 24 ++++++++++++-- 2 files changed, 75 insertions(+), 3 deletions(-) diff --git a/src/__tests__/eventQueue.test.ts b/src/__tests__/eventQueue.test.ts index 7621c51..dcd7dcf 100644 --- a/src/__tests__/eventQueue.test.ts +++ b/src/__tests__/eventQueue.test.ts @@ -831,6 +831,60 @@ describe("EventQueue", () => { }); }); + describe("cleanup robustness", () => { + it("still settles, and empties the queue, if teardown throws", async () => { + const { AppState } = jest.requireMock("react-native") as { + AppState: { addEventListener: jest.Mock }; + }; + AppState.addEventListener.mockReturnValueOnce({ + remove: () => { + throw new Error("native module gone"); + }, + }); + + const queue = makeQueue(); + const callback = jest.fn(); + await queue.enqueue(makeEvent(1), callback); + await settle(); + + // Must resolve rather than reject, and must not poison later calls. + await expect(queue.cleanup()).resolves.toBeUndefined(); + await expect(queue.cleanup()).resolves.toBeUndefined(); + + fetchMock.mockClear(); + await queue.flush(); + await settle(); + expect(fetchMock).not.toHaveBeenCalled(); + }); + + it("does not invoke a flush callback after teardown", async () => { + let releaseStall: (value: { ok: boolean; status: number }) => void; + fetchMock.mockReturnValueOnce( + new Promise((resolve) => { + releaseStall = resolve; + }) + ); + + const queue = makeQueue(); + await queue.enqueue(makeEvent(1)); + await settle(); + + // A consumer flush queued behind the stalled one. + const flushCallback = jest.fn(); + const racing = queue.flush(flushCallback); + + await queue.cleanup(); + expect(flushCallback).not.toHaveBeenCalled(); + + // The stalled send settles long after teardown; the queued flush then + // resumes and must not call back into the instance. + releaseStall!({ ok: true, status: 200 }); + await racing; + await settle(); + expect(flushCallback).not.toHaveBeenCalled(); + }, 20_000); + }); + describe("enqueue racing cleanup", () => { it("drops an event whose hashing was still pending when cleanup ran", async () => { const queue = makeQueue(); diff --git a/src/lib/event/EventQueue.ts b/src/lib/event/EventQueue.ts index bbe1a9e..54da597 100644 --- a/src/lib/event/EventQueue.ts +++ b/src/lib/event/EventQueue.ts @@ -368,8 +368,11 @@ export class EventQueue implements IEventQueue { // before cleanup() had even observed that flush finishing — delivering // events after teardown resolved. if (this.closed && !duringCleanup) { + // Deliberately no callback: this resumes only once the flush ahead + // settles, which can be long after cleanup() resolved, and nothing may + // call back into a torn-down instance. The returned promise still + // resolves, so an awaiting caller is never left hanging. logger.debug("EventQueue: Abandoning flush that outlived cleanup"); - safeCall(callback); return; } @@ -579,7 +582,16 @@ export class EventQueue implements IEventQueue { // Teardown is idempotent: a caller that asks twice joins the run already // under way rather than starting a competing one. if (!this.cleanupPromise) { - this.cleanupPromise = this.runCleanup(); + // Teardown is best-effort and must always settle: a rejection here would + // be memoised, so every later cleanup() would return the same rejected + // promise and never retry — and the provider, which awaits the pending + // cleanup before building a replacement, would reject with it forever. + this.cleanupPromise = this.runCleanup().catch((error) => { + logger.error("EventQueue: Cleanup failed", error); + // Whatever failed, the instance is going away: make sure nothing is + // left queued for a straggling flush to pick up. + this.clear(); + }); } return this.cleanupPromise; } @@ -597,7 +609,13 @@ export class EventQueue implements IEventQueue { // outlives cleanup(). The closed check in the handler covers the same // window; removing the listener means the event never reaches it at all. if (this.appStateSubscription) { - this.appStateSubscription.remove(); + try { + this.appStateSubscription.remove(); + } catch (error) { + // A native-module failure here must not abort teardown before the + // queue has been drained or invalidated below. + logger.error("EventQueue: Failed to remove AppState listener", error); + } this.appStateSubscription = null; }