diff --git a/apps/basket/src/lib/event-service.delivery.test.ts b/apps/basket/src/lib/event-service.delivery.test.ts index a35edf5be..303b59aeb 100644 --- a/apps/basket/src/lib/event-service.delivery.test.ts +++ b/apps/basket/src/lib/event-service.delivery.test.ts @@ -181,6 +181,82 @@ describe("event-service producer handoff", () => { expect(mockMarkDuplicateReservationDelivered).toHaveBeenCalledOnce(); }); + test("reserves track events only after enrichment completes", async () => { + let resolveGeo!: (value: { + anonymizedIP: string; + city: string; + country: string; + region: string; + }) => void; + mockGetGeo.mockImplementationOnce( + () => + new Promise((resolve) => { + resolveGeo = resolve; + }) + ); + + const pending = insertTrackEvent( + { eventId: "evt_1", name: "pageview", path: "/" }, + "ws_1", + "Mozilla/5.0", + "1.2.3.4", + new Request("https://basket.example/px.jpg") + ); + + expect(mockGetGeo).toHaveBeenCalledOnce(); + expect(mockReserveDuplicate).not.toHaveBeenCalled(); + resolveGeo({ + anonymizedIP: "1.2.3.0", + city: "San Francisco", + country: "US", + region: "CA", + }); + await pending; + + expect(mockReserveDuplicate.mock.invocationCallOrder[0]).toBeLessThan( + mockSend.mock.invocationCallOrder[0] as number + ); + }); + + test("reserves outgoing links only after GeoIP enrichment completes", async () => { + let resolveGeo!: (value: { + anonymizedIP: string; + city: string; + country: string; + region: string; + }) => void; + mockGetGeo.mockImplementationOnce( + () => + new Promise((resolve) => { + resolveGeo = resolve; + }) + ); + + const pending = insertOutgoingLink( + { + anonymizeVisitorIds: "auto", + eventId: "evt_link_1", + href: "https://external.example", + }, + "ws_1", + new Request("https://basket.example/px.jpg") + ); + + expect(mockGetGeo).toHaveBeenCalledOnce(); + expect(mockReserveDuplicate).not.toHaveBeenCalled(); + resolveGeo({ + anonymizedIP: "1.2.3.0", + city: "San Francisco", + country: "US", + region: "CA", + }); + await pending; + + expect(mockReserveDuplicate.mock.invocationCallOrder[0]).toBeLessThan( + mockSend.mock.invocationCallOrder[0] as number + ); + }); + test("propagates outgoing-link producer admission failures", async () => { const error = new Error("buffer full"); mockRunPromise.mockRejectedValueOnce(error); diff --git a/apps/basket/src/lib/event-service.ts b/apps/basket/src/lib/event-service.ts index b0abb5d4d..31259f3ef 100644 --- a/apps/basket/src/lib/event-service.ts +++ b/apps/basket/src/lib/event-service.ts @@ -16,6 +16,7 @@ import { reserveDuplicateBatch, shouldAnonymizeVisitorIds, } from "@lib/security"; +import { deliveryUnavailable } from "@lib/structured-errors"; import { record } from "@lib/tracing"; import { extractTrustedClientIp, getGeo } from "@utils/ip-geo"; import { parseUserAgent } from "@utils/user-agent"; @@ -27,7 +28,6 @@ import { validateSessionId, } from "@utils/validation"; import { randomUUIDv7 } from "bun"; -import { createError } from "evlog"; import { useLogger } from "evlog/elysia"; import { createHash } from "node:crypto"; @@ -127,17 +127,6 @@ export function stableBatchDeliveryId( .digest("hex"); } -function deliveryUnavailable(cause: unknown) { - return createError({ - code: "basket.DELIVERY_UNAVAILABLE", - message: "Analytics delivery temporarily unavailable", - status: 503, - why: "Databuddy could not durably accept the event.", - fix: "Retry the same event after a short delay.", - cause: cause instanceof Error ? cause : new Error(String(cause)), - }); -} - function directEventIdentity(eventId: unknown, generateFn: () => string) { const sourceEventId = typeof eventId === "string" && eventId.trim() @@ -249,20 +238,7 @@ export function insertTrackEvent( ); const deliveryId = stableAnalyticsEventId(clientId, "track", sourceEventId); - const reservation = await reserveDuplicate( - deliveryId, - "track", - storedEventId - ); - if (reservation.duplicate) { - return; - } - if (reservation.retryable) { - throw deliveryUnavailable( - new Error("A concurrent attempt owns this analytics event") - ); - } - + let trackEvent: EventsInsert; try { const geoData = await getGeo(ip, request); const trustedCountry = extractTrustedClientIp(request) @@ -298,7 +274,7 @@ export function insertTrackEvent( const now = Date.now(); - const trackEvent = buildTrackEvent(trackData, { + trackEvent = buildTrackEvent(trackData, { clientId, eventId: sourceEventId, anonymousId, @@ -306,7 +282,25 @@ export function insertTrackEvent( ua, now, }); + } catch (error) { + throw deliveryUnavailable(error); + } + + const reservation = await reserveDuplicate( + deliveryId, + "track", + storedEventId + ); + if (reservation.duplicate) { + return; + } + if (reservation.retryable) { + throw deliveryUnavailable( + new Error("A concurrent attempt owns this analytics event") + ); + } + try { await runPromise( send("analytics-events", trackEvent, undefined, { allowDirectFallback: reservation.ambiguous !== true, @@ -338,20 +332,7 @@ export function insertOutgoingLink( "outgoing_link", sourceEventId ); - const reservation = await reserveDuplicate( - deliveryId, - "outgoing_link", - storedEventId - ); - if (reservation.duplicate) { - return; - } - if (reservation.retryable) { - throw deliveryUnavailable( - new Error("A concurrent attempt owns this analytics event") - ); - } - + let outgoingLinkEvent: OutgoingLinksInsert; try { log.set({ event: { @@ -374,7 +355,7 @@ export function insertOutgoingLink( ); const salt = anonymizeVisitorIds ? await getDailySalt() : undefined; - const outgoingLinkEvent: OutgoingLinksInsert = { + outgoingLinkEvent = { id: deliveryId, client_id: clientId, anonymous_id: applyVisitorIdPrivacy( @@ -391,7 +372,25 @@ export function insertOutgoingLink( timestamp: typeof linkData.timestamp === "number" ? linkData.timestamp : now, }; + } catch (error) { + throw deliveryUnavailable(error); + } + + const reservation = await reserveDuplicate( + deliveryId, + "outgoing_link", + storedEventId + ); + if (reservation.duplicate) { + return; + } + if (reservation.retryable) { + throw deliveryUnavailable( + new Error("A concurrent attempt owns this analytics event") + ); + } + try { await runPromise( send("analytics-outgoing-links", outgoingLinkEvent, undefined, { allowDirectFallback: reservation.ambiguous !== true, diff --git a/apps/basket/src/lib/security.test.ts b/apps/basket/src/lib/security.test.ts index 361cc87a3..95c7ae6ca 100644 --- a/apps/basket/src/lib/security.test.ts +++ b/apps/basket/src/lib/security.test.ts @@ -174,7 +174,7 @@ describe("duplicate reservations", () => { "dedup:track:evt_1", expect.stringMatching(/^pending:/), "EX", - 120, + 30, "NX" ); }); @@ -248,7 +248,7 @@ describe("duplicate reservations", () => { expect.stringMatching(/^ambiguous-pending:\d+:/), "delivered", "ambiguous", - 120, + 30, expect.any(Number), "ambiguous-pending:", 86_400, @@ -322,11 +322,28 @@ describe("duplicate reservations", () => { "dedup:track:stable-delivery-id", expect.stringMatching(/^pending:/), "EX", - 120, + 30, "NX" ); }); + test("allows a later retry after a stale pending lease", async () => { + mockRedisSet.mockResolvedValueOnce(null).mockResolvedValueOnce("OK"); + mockRedisGet.mockResolvedValue("pending:crashed-owner"); + + expect(await reserveDuplicate("evt_1", "track")).toEqual({ + duplicate: false, + retryable: true, + }); + + expect(await reserveDuplicate("evt_1", "track")).toMatchObject({ + deliveredTtl: 86_400, + duplicate: false, + key: "dedup:track:evt_1", + token: expect.stringMatching(/^pending:/), + }); + }); + test("retries a Redis error before acquiring the reservation", async () => { mockRedisSet .mockRejectedValueOnce(new Error("stale connection")) diff --git a/apps/basket/src/lib/security.ts b/apps/basket/src/lib/security.ts index 2411a21bf..5e34fed76 100644 --- a/apps/basket/src/lib/security.ts +++ b/apps/basket/src/lib/security.ts @@ -8,7 +8,10 @@ import { useLogger } from "evlog/elysia"; const EXIT_EVENT_TTL = 172_800; const STANDARD_EVENT_TTL = 86_400; -const PENDING_DEDUP_TTL = 120; +// Reservations begin immediately before the bounded producer handoff. This +// exceeds the 20-second Railway shutdown budget while letting a crashed owner +// expire before client retries are suppressed for minutes. +const PENDING_DEDUP_TTL = 30; const DEDUP_RETRY_DELAY_MS = 25; export const DEDUP_RESERVATION_TIMEOUT_MS = 750; const DEDUP_FAILURE_COOLDOWN_MS = 5000; @@ -393,7 +396,7 @@ export function reserveDuplicate( if (error instanceof DeduplicationDeadlineError) { // Redis commands cannot be cancelled. If SET NX succeeds after the // caller returns retryable, conditionally reconcile only this attempt's - // token so it cannot strand an ownerless 120-second reservation. + // token so it cannot strand an ownerless pending reservation. reservationOperation .then(async (state) => { if (state === "ambiguous-acquired") { diff --git a/apps/basket/src/lib/structured-errors.test.ts b/apps/basket/src/lib/structured-errors.test.ts index 2b87c36ba..2fa714aaa 100644 --- a/apps/basket/src/lib/structured-errors.test.ts +++ b/apps/basket/src/lib/structured-errors.test.ts @@ -4,6 +4,7 @@ import { basketErrors, buildBasketErrorPayload, createIngestSchemaValidationError, + deliveryUnavailable, isIngestSchemaValidationError, rethrowOrWrap, } from "./structured-errors"; @@ -86,6 +87,17 @@ describe("IngestSchemaValidationError", () => { }); }); +describe("deliveryUnavailable", () => { + test("creates a retryable structured 503", () => { + const error = deliveryUnavailable(new Error("Redis unavailable")); + + expect(error).toMatchObject({ + code: "basket.DELIVERY_UNAVAILABLE", + status: 503, + }); + }); +}); + // ── rethrowOrWrap ── describe("rethrowOrWrap", () => { diff --git a/apps/basket/src/lib/structured-errors.ts b/apps/basket/src/lib/structured-errors.ts index c6bc3e819..810b5c99b 100644 --- a/apps/basket/src/lib/structured-errors.ts +++ b/apps/basket/src/lib/structured-errors.ts @@ -227,6 +227,21 @@ export function createIngestSchemaValidationError( return Object.assign(err, { issues }); } +/** + * A request must not report success while its telemetry could not be durably + * admitted. Callers return this 503 so SDKs and queueing clients retry. + */ +export function deliveryUnavailable(cause: unknown) { + return createError({ + code: "basket.DELIVERY_UNAVAILABLE", + message: "Analytics delivery temporarily unavailable", + status: 503, + why: "Databuddy could not durably accept the event.", + fix: "Retry the same event after a short delay.", + cause: cause instanceof Error ? cause : new Error(String(cause)), + }); +} + export function isIngestSchemaValidationError( error: unknown ): error is IngestSchemaValidationError { diff --git a/apps/basket/src/routes/basket.ts b/apps/basket/src/routes/basket.ts index 34e7d95ec..a9c236fb0 100644 --- a/apps/basket/src/routes/basket.ts +++ b/apps/basket/src/routes/basket.ts @@ -42,6 +42,7 @@ import { import { basketErrors, createIngestSchemaValidationError, + deliveryUnavailable, rethrowOrWrap, } from "@lib/structured-errors"; import { record } from "@lib/tracing"; @@ -518,7 +519,10 @@ const app = new Elysia() }; for (const event of body) { - const eventType = event.type || "track"; + const isEventObject = + event !== null && typeof event === "object" && !Array.isArray(event); + const eventType = isEventObject ? event.type || "track" : "track"; + const eventId = isEventObject ? event.eventId : undefined; try { if (eventType === "track") { @@ -547,7 +551,7 @@ const app = new Elysia() batchSchemaItemFailure( parseResult.error.issues, eventType, - event.eventId + eventId ) ); continue; @@ -592,7 +596,7 @@ const app = new Elysia() batchSchemaItemFailure( parseResult.error.issues, eventType, - event.eventId + eventId ) ); continue; @@ -621,13 +625,17 @@ const app = new Elysia() }); } } catch (error) { - log.error(error instanceof Error ? error : new Error(String(error))); - results.push({ - status: "error", - message: "Processing failed", - code: "EVENT_PROCESSING_FAILED", - eventType, - }); + if ( + error instanceof EvlogError && + error.status === 503 && + error.code === "basket.DELIVERY_UNAVAILABLE" + ) { + throw error; + } + const processingError = + error instanceof Error ? error : new Error(String(error)); + log.error(processingError); + throw deliveryUnavailable(processingError); } } diff --git a/apps/basket/src/routes/integration.test.ts b/apps/basket/src/routes/integration.test.ts index 4de96b340..4a6d3957f 100644 --- a/apps/basket/src/routes/integration.test.ts +++ b/apps/basket/src/routes/integration.test.ts @@ -13,6 +13,7 @@ const { mockInsertIndividualVitals, mockInsertErrorSpans, mockInsertCustomEvents, + mockGetGeo, mockCheckAutumnUsage, mockGetApiKeyFromHeader, mockHasKeyScope, @@ -63,6 +64,14 @@ const { mockInsertIndividualVitals: vi.fn(() => Promise.resolve()), mockInsertErrorSpans: vi.fn(() => Promise.resolve()), mockInsertCustomEvents: vi.fn(() => Promise.resolve()), + mockGetGeo: vi.fn(() => + Promise.resolve({ + anonymizedIP: "abc123", + country: "US", + region: "CA", + city: "SF", + }) + ), mockCheckAutumnUsage: vi.fn(() => Promise.resolve({ allowed: true })), mockGetApiKeyFromHeader: vi.fn(() => Promise.resolve(defaultApiKey)), mockHasKeyScope: vi.fn(() => true), @@ -80,6 +89,7 @@ vi.mock("evlog/elysia", () => ({ vi.mock("@lib/tracing", () => ({ record: (_n: string, fn: Function) => Promise.resolve().then(() => fn()), captureError: noop, + mergeWideEvent: noop, })); vi.mock("@lib/request-validation", () => ({ @@ -114,14 +124,7 @@ vi.mock("@lib/security", () => ({ })); vi.mock("@utils/ip-geo", () => ({ - getGeo: vi.fn(() => - Promise.resolve({ - anonymizedIP: "abc123", - country: "US", - region: "CA", - city: "SF", - }) - ), + getGeo: mockGetGeo, extractIpFromRequest: vi.fn(() => "1.2.3.4"), extractTrustedClientIp: vi.fn(() => "1.2.3.4"), getVisitorCountryForAutoMode: vi.fn((events: Array<{ anonymizeVisitorIds?: unknown }>) => @@ -578,6 +581,75 @@ describe("POST /batch", () => { expect(await json(res)).toMatchObject({ batch: true, processed: 1 }); }); + test("returns 503 instead of accepting an event that could not be prepared", async () => { + mockInsertTrackEventsBatch.mockClear(); + mockInsertOutgoingLinksBatch.mockClear(); + mockGetGeo + .mockResolvedValueOnce({ + anonymizedIP: "abc123", + country: "US", + region: "CA", + city: "SF", + }) + .mockRejectedValueOnce(new Error("GeoIP unavailable")); + + const result = await post(basketApp, "/batch", [ + { + type: "track", + eventId: "evt_1", + name: "pageview", + path: "https://example.com/a", + }, + { + type: "track", + eventId: "evt_2", + name: "click", + path: "https://example.com/b", + }, + ]); + + expect(result.status).toBe(503); + expect(await json(result)).toMatchObject({ + code: "basket.DELIVERY_UNAVAILABLE", + retryable: true, + }); + expect(mockInsertTrackEventsBatch).not.toHaveBeenCalled(); + expect(mockInsertOutgoingLinksBatch).not.toHaveBeenCalled(); + }); + + test("keeps malformed items as schema failures without dropping valid events", async () => { + mockInsertTrackEventsBatch.mockClear(); + mockInsertOutgoingLinksBatch.mockClear(); + + const result = await post(basketApp, "/batch", [ + { + type: "track", + eventId: "evt_1", + name: "pageview", + path: "https://example.com/a", + }, + null, + ]); + + expect(result.status).toBe(200); + expect(await json(result)).toMatchObject({ + status: "partial", + batch: true, + processed: 2, + batched: { track: 1, outgoing_link: 0 }, + results: [ + { status: "success", type: "track", eventId: "evt_1" }, + { + status: "error", + code: "INVALID_EVENT_SCHEMA", + eventType: "track", + }, + ], + }); + expect(mockInsertTrackEventsBatch).toHaveBeenCalledOnce(); + expect(mockInsertOutgoingLinksBatch).toHaveBeenCalledOnce(); + }); + test("not an array → 400", async () => { const res = await post(basketApp, "/batch", { not: "array" }); expect(res.status).toBe(400); @@ -594,7 +666,7 @@ describe("POST /batch", () => { expect(res.status).toBe(400); }); - test("mixed valid + unknown types → partial results", async () => { + test("mixed valid + invalid types → partial results", async () => { const res = await post(basketApp, "/batch", [ { type: "track", @@ -602,7 +674,7 @@ describe("POST /batch", () => { name: "pageview", path: "https://example.com/a", }, - { type: "bogus_type" }, + { type: 1 }, ]); expect(res.status).toBe(200); const body = await json(res); diff --git a/packages/ai/src/ai/agents/cache.test.ts b/packages/ai/src/ai/agents/cache.test.ts index ba742d2b3..03fe4864c 100644 --- a/packages/ai/src/ai/agents/cache.test.ts +++ b/packages/ai/src/ai/agents/cache.test.ts @@ -169,6 +169,7 @@ vi.mock("@databuddy/redis", () => ({ redis: mockRedisClient, setActiveStream: vi.fn(async () => undefined), setCachedLink: vi.fn(async () => undefined), + setCachedLinkIfAbsent: vi.fn(async () => true), setCachedLinkNotFound: vi.fn(async () => undefined), shouldRecordClick: vi.fn(async () => true), shutdownRedis: vi.fn(async () => undefined),