test(evidence): cover AUD0 outbox delivery and failure semantics
This commit is contained in:
@@ -0,0 +1,155 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
|
||||
import type { Aud0Client, Aud0Event } from "@/src/clients/aud0-client";
|
||||
import {
|
||||
Aud0OutboxDispatcher,
|
||||
openLineageIdempotencyKey,
|
||||
outboxId,
|
||||
type Aud0OutboxRecord,
|
||||
type Aud0OutboxRepository,
|
||||
type Aud0OutboxStats,
|
||||
} from "@/src/evidence/aud0-outbox";
|
||||
import type { OpenLineageRunEvent } from "@/src/lineage/openlineage";
|
||||
import { SiaxError } from "@/src/platform/errors";
|
||||
|
||||
const event: Aud0Event = {
|
||||
event_id: "evt-1",
|
||||
event_type: "0pt.test",
|
||||
tenant_id: "tenant-a",
|
||||
project_id: "project-a",
|
||||
actor_id: "tester",
|
||||
app_id: "0pt",
|
||||
capability_id: "0pt.test.v1",
|
||||
resource_type: "test",
|
||||
resource_id: "r1",
|
||||
risk_level: "L1",
|
||||
created_at: "2026-09-07T12:00:00.000Z",
|
||||
idempotency_key: "idem-1",
|
||||
};
|
||||
|
||||
const ol: OpenLineageRunEvent = {
|
||||
eventType: "COMPLETE",
|
||||
eventTime: "2026-09-07T12:00:00.000Z",
|
||||
producer: "https://git.siax.io/sax3l/0pt",
|
||||
run: { runId: "run-1" },
|
||||
job: { namespace: "0pt", name: "0pt.test.v1" },
|
||||
inputs: [],
|
||||
outputs: [],
|
||||
};
|
||||
|
||||
class FakeRepo implements Aud0OutboxRepository {
|
||||
claimed: Aud0OutboxRecord[] = [];
|
||||
delivered: string[] = [];
|
||||
dead: Array<{ id: string; error: string }> = [];
|
||||
retried: Array<{ id: string; error: string; delayMs: number }> = [];
|
||||
|
||||
async enqueueEvent(): Promise<Aud0OutboxRecord> { throw new Error("unused"); }
|
||||
async enqueueOpenLineage(): Promise<Aud0OutboxRecord> { throw new Error("unused"); }
|
||||
async claimPending(): Promise<Aud0OutboxRecord[]> { return this.claimed; }
|
||||
async markDelivered(id: string): Promise<void> { this.delivered.push(id); }
|
||||
async markRetry(id: string, _worker: string, error: string, delayMs: number): Promise<void> {
|
||||
this.retried.push({ id, error, delayMs });
|
||||
}
|
||||
async markDead(id: string, _worker: string, error: string): Promise<void> {
|
||||
this.dead.push({ id, error });
|
||||
}
|
||||
async stats(): Promise<Aud0OutboxStats> { return { pending: this.claimed.length, delivered: 0, dead: 0 }; }
|
||||
}
|
||||
|
||||
function row(
|
||||
kind: "event" | "openlineage",
|
||||
payload: Aud0Event | OpenLineageRunEvent,
|
||||
overrides: Partial<Aud0OutboxRecord> = {},
|
||||
): Aud0OutboxRecord {
|
||||
return {
|
||||
outbox_id: "obx-1",
|
||||
tenant_id: "tenant-a",
|
||||
app_id: "0pt",
|
||||
evidence_kind: kind,
|
||||
idempotency_key: "idem-1",
|
||||
payload,
|
||||
status: "pending",
|
||||
attempts: 1,
|
||||
next_attempt_at: "2026-09-07T12:00:00.000Z",
|
||||
created_at: "2026-09-07T12:00:00.000Z",
|
||||
updated_at: "2026-09-07T12:00:00.000Z",
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
function client(impl?: Partial<Aud0Client>): Aud0Client {
|
||||
return {
|
||||
async ingestEvent(input) { return input; },
|
||||
async ingestOpenLineageEvent(tenantId, _appId, input) {
|
||||
return { tenant_id: tenantId, run_id: input.run.runId, job: input.job.name, event_id: "aud0-1" };
|
||||
},
|
||||
async ingestLineageEvent(input) { return input; },
|
||||
...impl,
|
||||
};
|
||||
}
|
||||
|
||||
describe("AUD0 durable outbox", () => {
|
||||
it("derives stable tenant-bound outbox ids", () => {
|
||||
expect(outboxId("tenant-a", "event", "k1")).toBe(outboxId("tenant-a", "event", "k1"));
|
||||
expect(outboxId("tenant-a", "event", "k1")).not.toBe(outboxId("tenant-b", "event", "k1"));
|
||||
expect(outboxId("tenant-a", "event", "k1")).not.toBe(outboxId("tenant-a", "openlineage", "k1"));
|
||||
});
|
||||
|
||||
it("uses logical OpenLineage run+eventType identity rather than delivery time", () => {
|
||||
expect(openLineageIdempotencyKey(ol)).toBe("ol:run-1:COMPLETE");
|
||||
expect(openLineageIdempotencyKey({ ...ol, eventTime: "2030-01-01T00:00:00Z" })).toBe("ol:run-1:COMPLETE");
|
||||
});
|
||||
|
||||
it("marks a successfully delivered generic event delivered", async () => {
|
||||
const repo = new FakeRepo();
|
||||
repo.claimed = [row("event", event)];
|
||||
let calls = 0;
|
||||
const dispatcher = new Aud0OutboxDispatcher(repo, client({
|
||||
async ingestEvent(input) { calls += 1; return input; },
|
||||
}), "worker-a");
|
||||
|
||||
expect(await dispatcher.sweep()).toEqual({ claimed: 1, delivered: 1, dead: 0, retried: 0 });
|
||||
expect(calls).toBe(1);
|
||||
expect(repo.delivered).toEqual(["obx-1"]);
|
||||
});
|
||||
|
||||
it("retries network/dependency failures with bounded exponential backoff", async () => {
|
||||
const repo = new FakeRepo();
|
||||
repo.claimed = [row("openlineage", ol, { attempts: 4 })];
|
||||
const dispatcher = new Aud0OutboxDispatcher(repo, client({
|
||||
async ingestOpenLineageEvent() {
|
||||
throw new SiaxError("DEPENDENCY_UNAVAILABLE", "aud0", "offline");
|
||||
},
|
||||
}), "worker-a");
|
||||
|
||||
expect(await dispatcher.sweep()).toEqual({ claimed: 1, delivered: 0, dead: 0, retried: 1 });
|
||||
expect(repo.retried[0].delayMs).toBe(8_000);
|
||||
expect(repo.retried[0].error).toBe("DEPENDENCY_UNAVAILABLE:aud0");
|
||||
});
|
||||
|
||||
it("parks permanent 4xx-style failures dead instead of poisoning the queue", async () => {
|
||||
const repo = new FakeRepo();
|
||||
repo.claimed = [row("event", event)];
|
||||
const dispatcher = new Aud0OutboxDispatcher(repo, client({
|
||||
async ingestEvent() {
|
||||
throw new SiaxError("INVALID_REQUEST", "aud0", "bad event");
|
||||
},
|
||||
}), "worker-a");
|
||||
|
||||
expect(await dispatcher.sweep()).toEqual({ claimed: 1, delivered: 0, dead: 1, retried: 0 });
|
||||
expect(repo.dead).toEqual([{ id: "obx-1", error: "INVALID_REQUEST:aud0" }]);
|
||||
});
|
||||
|
||||
it("fails closed on an immutable tenant/app envelope mismatch before calling AUD0", async () => {
|
||||
const repo = new FakeRepo();
|
||||
repo.claimed = [row("event", { ...event, tenant_id: "tenant-b" })];
|
||||
let calls = 0;
|
||||
const dispatcher = new Aud0OutboxDispatcher(repo, client({
|
||||
async ingestEvent(input) { calls += 1; return input; },
|
||||
}), "worker-a");
|
||||
|
||||
expect(await dispatcher.sweep()).toEqual({ claimed: 1, delivered: 0, dead: 1, retried: 0 });
|
||||
expect(calls).toBe(0);
|
||||
expect(repo.dead[0].error).toBe("PERMANENT_FAILURE:0pt.aud0_outbox");
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user