Skip to content

fix(basket): preserve retries after interrupted admission #586

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
izadoesdev merged 4 commits into from
Aug 2, 2026
Merged

fix(basket): preserve retries after interrupted admission #586

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
6c737d4
fix(basket): preserve retries after interrupted admission
izadoesdev Aug 2, 2026
9c3ff98
fix(basket): reserve events at delivery handoff
izadoesdev Aug 2, 2026
188d967
fix(ai): keep Redis mock compatible
izadoesdev Aug 2, 2026
49ca094
Merge branch 'staging' into codex/basket-recovery-staging
izadoesdev Aug 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
76 changes: 76 additions & 0 deletions apps/basket/src/lib/event-service.delivery.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
83 changes: 41 additions & 42 deletions apps/basket/src/lib/event-service.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand All @@ -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";

Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -298,15 +274,33 @@ export function insertTrackEvent(

const now = Date.now();

const trackEvent = buildTrackEvent(trackData, {
trackEvent = buildTrackEvent(trackData, {
clientId,
eventId: sourceEventId,
anonymousId,
geo: geoData,
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,
Expand Down Expand Up @@ -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: {
Expand All @@ -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(
Expand All @@ -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,
Expand Down
23 changes: 20 additions & 3 deletions apps/basket/src/lib/security.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -174,7 +174,7 @@ describe("duplicate reservations", () => {
"dedup:track:evt_1",
expect.stringMatching(/^pending:/),
"EX",
120,
30,
"NX"
);
});
Expand Down Expand Up @@ -248,7 +248,7 @@ describe("duplicate reservations", () => {
expect.stringMatching(/^ambiguous-pending:\d+:/),
"delivered",
"ambiguous",
120,
30,
expect.any(Number),
"ambiguous-pending:",
86_400,
Expand Down Expand Up @@ -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"))
Expand Down
7 changes: 5 additions & 2 deletions apps/basket/src/lib/security.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

@greptile-apps greptile-apps Bot Aug 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Pending lease expires during admission

If enrichment or producer delivery remains in flight for more than 30 seconds, the pending Redis key expires before the original owner marks it delivered. A retry can then reserve the same event ID and publish it again, causing duplicate analytics rows while the original owner's token no longer matches.

Knowledge Base Used: Basket Ingestion Flow

const DEDUP_RETRY_DELAY_MS = 25;
export const DEDUP_RESERVATION_TIMEOUT_MS = 750;
const DEDUP_FAILURE_COOLDOWN_MS = 5000;
Expand Down Expand Up @@ -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") {
Expand Down
12 changes: 12 additions & 0 deletions apps/basket/src/lib/structured-errors.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import {
basketErrors,
buildBasketErrorPayload,
createIngestSchemaValidationError,
deliveryUnavailable,
isIngestSchemaValidationError,
rethrowOrWrap,
} from "./structured-errors";
Expand Down Expand Up @@ -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", () => {
Expand Down
15 changes: 15 additions & 0 deletions apps/basket/src/lib/structured-errors.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
Loading
Loading