refactor(supervisor): op + kind first-class on State
This commit is contained in:
@@ -259,6 +259,8 @@ class ManagedSupervisor {
|
||||
await runWideEvent(
|
||||
{
|
||||
...this.wideEventOpts,
|
||||
op: "dequeue",
|
||||
kind: "inbound",
|
||||
traceparent,
|
||||
setup: (state) => {
|
||||
setMeta(state, "run_id", message.run.friendlyId);
|
||||
@@ -269,8 +271,6 @@ class ManagedSupervisor {
|
||||
setMeta(state, "deployment_id", message.deployment.friendlyId);
|
||||
}
|
||||
setMeta(state, "machine_preset", message.run.machine.name);
|
||||
state.extras.op = "dequeue";
|
||||
state.extras.kind = "inbound";
|
||||
state.extras.iteration = "dequeue";
|
||||
state.extras.dequeue_response_ms = dequeueResponseMs;
|
||||
state.extras.polling_interval_ms = pollingIntervalMs;
|
||||
|
||||
@@ -76,9 +76,9 @@ export class ComputeSnapshotService {
|
||||
this.timerWheel.submit(runFriendlyId, data);
|
||||
emitOneShot({
|
||||
...this.wideEventOpts,
|
||||
op: "snapshot.schedule",
|
||||
kind: "event",
|
||||
populate: (state) => {
|
||||
state.extras.op = "snapshot.schedule";
|
||||
state.extras.kind = "event";
|
||||
state.meta.run_id = runFriendlyId;
|
||||
state.meta.snapshot_id = data.snapshotFriendlyId;
|
||||
state.extras.runner_id = data.runnerId;
|
||||
@@ -98,9 +98,9 @@ export class ComputeSnapshotService {
|
||||
if (cancelled) {
|
||||
emitOneShot({
|
||||
...this.wideEventOpts,
|
||||
op: "snapshot.canceled",
|
||||
kind: "event",
|
||||
populate: (state) => {
|
||||
state.extras.op = "snapshot.canceled";
|
||||
state.extras.kind = "event";
|
||||
state.meta.run_id = runFriendlyId;
|
||||
},
|
||||
});
|
||||
@@ -121,7 +121,6 @@ export class ComputeSnapshotService {
|
||||
// become extras/meta on the same wide event - no nested emission.
|
||||
const state = fromContext();
|
||||
if (state) {
|
||||
state.extras.op = "snapshot.callback";
|
||||
state.extras["snapshot.status"] = body.status;
|
||||
if (body.instance_id) state.extras["snapshot.instance_id"] = body.instance_id;
|
||||
if (body.duration_ms !== undefined) state.extras["snapshot.duration_ms"] = body.duration_ms;
|
||||
@@ -247,9 +246,9 @@ export class ComputeSnapshotService {
|
||||
await runWideEvent(
|
||||
{
|
||||
...this.wideEventOpts,
|
||||
op: "snapshot.dispatch",
|
||||
kind: "scheduled",
|
||||
setup: (state) => {
|
||||
state.extras.op = "snapshot.dispatch";
|
||||
state.extras.kind = "scheduled";
|
||||
state.meta.run_id = snapshot.runFriendlyId;
|
||||
state.meta.snapshot_id = snapshot.snapshotFriendlyId;
|
||||
state.extras.runner_id = snapshot.runnerId;
|
||||
|
||||
@@ -27,6 +27,9 @@ export function emit(state: State): void {
|
||||
appendIfSet(out, "region", state.region);
|
||||
appendIfSet(out, "node_id", state.nodeId);
|
||||
|
||||
appendIfSet(out, "op", state.op);
|
||||
appendIfSet(out, "kind", state.kind);
|
||||
|
||||
out.ok = state.ok;
|
||||
if (state.statusCode !== 0) out.status = state.statusCode;
|
||||
out.duration_ms = state.durationMs;
|
||||
|
||||
@@ -20,7 +20,7 @@ describe("runWideEvent", () => {
|
||||
it("emits one event with ok=true when no statusCode is set", async () => {
|
||||
const lines = await captureStdout(async () => {
|
||||
await runWideEvent(
|
||||
{ service: "supervisor", env: {}, enabled: true, route: "/x", method: "POST" },
|
||||
{ service: "supervisor", env: {}, enabled: true, op: "test", route: "/x", method: "POST" },
|
||||
async () => undefined
|
||||
);
|
||||
});
|
||||
@@ -39,7 +39,7 @@ describe("runWideEvent", () => {
|
||||
it("derives ok from statusCode set via finalize", async () => {
|
||||
const lines = await captureStdout(async () => {
|
||||
await runWideEvent(
|
||||
{ service: "supervisor", env: {}, enabled: true },
|
||||
{ service: "supervisor", env: {}, enabled: true, op: "test" },
|
||||
async () => undefined,
|
||||
(state) => {
|
||||
state.statusCode = 200;
|
||||
@@ -56,7 +56,7 @@ describe("runWideEvent", () => {
|
||||
it("treats 4xx as ok=false", async () => {
|
||||
const lines = await captureStdout(async () => {
|
||||
await runWideEvent(
|
||||
{ service: "supervisor", env: {}, enabled: true },
|
||||
{ service: "supervisor", env: {}, enabled: true, op: "test" },
|
||||
async () => undefined,
|
||||
(state) => {
|
||||
state.statusCode = 400;
|
||||
@@ -73,7 +73,7 @@ describe("runWideEvent", () => {
|
||||
it("emits ok=false with error.kind=internal on throw", async () => {
|
||||
const lines = await captureStdout(async () => {
|
||||
await runWideEvent(
|
||||
{ service: "supervisor", env: {}, enabled: true },
|
||||
{ service: "supervisor", env: {}, enabled: true, op: "test" },
|
||||
async () => {
|
||||
throw new Error("boom");
|
||||
}
|
||||
@@ -91,7 +91,7 @@ describe("runWideEvent", () => {
|
||||
it("threads state through AsyncLocalStorage", async () => {
|
||||
const lines = await captureStdout(async () => {
|
||||
await runWideEvent(
|
||||
{ service: "supervisor", env: {}, enabled: true },
|
||||
{ service: "supervisor", env: {}, enabled: true, op: "test" },
|
||||
async () => {
|
||||
setMeta(fromContext(), "run_id", "run_abc");
|
||||
}
|
||||
@@ -108,7 +108,7 @@ describe("runWideEvent", () => {
|
||||
const tp = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01";
|
||||
const lines = await captureStdout(async () => {
|
||||
await runWideEvent(
|
||||
{ service: "supervisor", env: {}, enabled: true, traceparent: tp },
|
||||
{ service: "supervisor", env: {}, enabled: true, op: "test", traceparent: tp },
|
||||
async () => undefined
|
||||
);
|
||||
});
|
||||
@@ -124,7 +124,7 @@ describe("runWideEvent", () => {
|
||||
{
|
||||
service: "supervisor",
|
||||
env: {},
|
||||
enabled: true,
|
||||
enabled: true, op: "test",
|
||||
setup: (state) => {
|
||||
state.meta.run_id = "run_abc";
|
||||
state.extras.iteration = "dequeue";
|
||||
@@ -144,7 +144,7 @@ describe("runWideEvent", () => {
|
||||
let seenState: ReturnType<typeof fromContext> = null;
|
||||
const lines = await captureStdout(async () => {
|
||||
await runWideEvent(
|
||||
{ service: "supervisor", env: {}, enabled: false },
|
||||
{ service: "supervisor", env: {}, enabled: false, op: "test" },
|
||||
async () => {
|
||||
seenState = fromContext();
|
||||
}
|
||||
@@ -159,7 +159,7 @@ describe("runWideEvent", () => {
|
||||
await Promise.all(
|
||||
["a", "b", "c"].map((tag) =>
|
||||
runWideEvent(
|
||||
{ service: "supervisor", env: {}, enabled: true },
|
||||
{ service: "supervisor", env: {}, enabled: true, op: "test" },
|
||||
async () => {
|
||||
const s = fromContext();
|
||||
if (!s) throw new Error("no state");
|
||||
@@ -182,7 +182,7 @@ describe("emitOneShot", () => {
|
||||
emitOneShot({
|
||||
service: "supervisor",
|
||||
env: {},
|
||||
enabled: true,
|
||||
enabled: true, op: "test",
|
||||
populate: (s) => {
|
||||
s.meta.run_id = "run_abc";
|
||||
s.extras.event = "run:start";
|
||||
@@ -200,7 +200,7 @@ describe("emitOneShot", () => {
|
||||
|
||||
it("emits nothing when disabled", async () => {
|
||||
const lines = await captureStdout(() => {
|
||||
emitOneShot({ service: "supervisor", env: {}, enabled: false });
|
||||
emitOneShot({ service: "supervisor", env: {}, enabled: false, op: "test" });
|
||||
});
|
||||
expect(lines).toHaveLength(0);
|
||||
});
|
||||
|
||||
@@ -18,6 +18,10 @@ export type WideEventOptions = {
|
||||
|
||||
/** Per-invocation options layered on top of `WideEventOptions`. */
|
||||
export type WideEventLifecycleOptions = WideEventOptions & {
|
||||
/** Operation discriminator (`instance.create`, `dequeue`, ...). Required. */
|
||||
op: string;
|
||||
/** Event shape: `inbound` | `outbound` | `event` | `scheduled`. Optional. */
|
||||
kind?: string;
|
||||
/** Route template (HTTP only) captured into `extras.route`. */
|
||||
route?: string;
|
||||
/** HTTP method captured into `extras.method`. */
|
||||
@@ -54,6 +58,8 @@ export async function runWideEvent<T>(
|
||||
env: opts.env,
|
||||
inboundRequestId: opts.inboundRequestId,
|
||||
traceparent: opts.traceparent,
|
||||
op: opts.op,
|
||||
kind: opts.kind,
|
||||
});
|
||||
if (opts.route) state.extras.route = opts.route;
|
||||
if (opts.method) state.extras.method = opts.method;
|
||||
@@ -94,6 +100,8 @@ export async function runWideEvent<T>(
|
||||
*/
|
||||
export function emitOneShot(
|
||||
opts: WideEventOptions & {
|
||||
op: string;
|
||||
kind?: string;
|
||||
traceparent?: string;
|
||||
populate?: (state: State) => void;
|
||||
}
|
||||
@@ -103,6 +111,8 @@ export function emitOneShot(
|
||||
service: opts.service,
|
||||
env: opts.env,
|
||||
traceparent: opts.traceparent,
|
||||
op: opts.op,
|
||||
kind: opts.kind,
|
||||
});
|
||||
if (opts.populate) opts.populate(state);
|
||||
state.ok = true;
|
||||
|
||||
@@ -37,6 +37,10 @@ export type NewStateOptions = {
|
||||
inboundRequestId?: string;
|
||||
/** Optional inbound W3C traceparent (HTTP header, queue message field). */
|
||||
traceparent?: string;
|
||||
/** Operation discriminator. Dotted `noun.verb`. Defaults to empty (set later). */
|
||||
op?: string;
|
||||
/** Event shape: `inbound` | `outbound` | `event` | `scheduled`. Defaults to empty. */
|
||||
kind?: string;
|
||||
};
|
||||
|
||||
/**
|
||||
@@ -62,6 +66,8 @@ export function newState(opts: NewStateOptions): State {
|
||||
commitSha: opts.env.commitSha,
|
||||
region: opts.env.region,
|
||||
nodeId: opts.env.nodeId,
|
||||
op: opts.op ?? "",
|
||||
kind: opts.kind ?? "",
|
||||
meta: {},
|
||||
phases: [],
|
||||
ok: false,
|
||||
|
||||
@@ -22,6 +22,22 @@ export type State = {
|
||||
region?: string;
|
||||
nodeId?: string;
|
||||
|
||||
/**
|
||||
* Operation discriminator. Dotted `noun.verb` (e.g. `instance.create`,
|
||||
* `snapshot.dispatch`). Low cardinality - bounded set per service, not
|
||||
* unbounded. Empty allowed during construction but expected to be set
|
||||
* before emit.
|
||||
*/
|
||||
op: string;
|
||||
|
||||
/**
|
||||
* Event shape. `inbound` for received requests, `outbound` for outgoing
|
||||
* calls, `event` for ambient occurrences with no meaningful duration,
|
||||
* `scheduled` for timer-driven work. Empty allowed; omitted from emit
|
||||
* when empty.
|
||||
*/
|
||||
kind: string;
|
||||
|
||||
// Caller-attached opaque metadata, flattened to `meta.<key>` on emit.
|
||||
meta: Record<string, string>;
|
||||
|
||||
|
||||
@@ -208,15 +208,13 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
|
||||
{
|
||||
...this.wideEventOpts,
|
||||
enabled,
|
||||
op,
|
||||
kind: "inbound",
|
||||
route,
|
||||
method,
|
||||
traceparent: this.headerValueFromRequest(ctx.req, "traceparent"),
|
||||
inboundRequestId: this.headerValueFromRequest(ctx.req, "x-request-id"),
|
||||
setup: (state) => {
|
||||
state.extras.op = op;
|
||||
state.extras.kind = "inbound";
|
||||
this.attachRouteMeta(state, ctx.params);
|
||||
},
|
||||
setup: (state) => this.attachRouteMeta(state, ctx.params),
|
||||
},
|
||||
fn,
|
||||
(state) => {
|
||||
@@ -693,9 +691,9 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
|
||||
) => {
|
||||
emitOneShot({
|
||||
...this.wideEventOpts,
|
||||
op: event === "run_connected" ? "socket.run.connected" : "socket.run.disconnected",
|
||||
kind: "event",
|
||||
populate: (state) => {
|
||||
state.extras.op = event === "run_connected" ? "socket.run.connected" : "socket.run.disconnected";
|
||||
state.extras.kind = "event";
|
||||
state.extras.event = event;
|
||||
setMeta(state, "run_id", friendlyId);
|
||||
if (socket.data.deploymentId) {
|
||||
|
||||
Reference in New Issue
Block a user