Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
172 changes: 172 additions & 0 deletions daemon/coordinator-lease.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,172 @@
/**
* Coordinator-backed lease client — drop-in replacement for daemon/leader.ts.
*
* Exports the same four functions with the same signatures and return shapes,
* so daemon/loop.ts switches by changing one import path. The arbiter moves
* from Neon `cc_node_leases` to the CommandCoordinator Durable Object
* (meta/coordinator.ts).
*
* Why: ADR-001 used Neon leases to elect a leader across chittymini-01..06.
* A DO is already a strongly-consistent singleton, so election is unnecessary —
* and the Neon dependency, which is cost-driven pressure, leaves this layer.
*
* Fails closed. With no coordinator configured this throws rather than
* degrading to an unarbitrated local decision, which would permit split-brain.
*
* @canonical-uri chittycanon://docs/architecture/chittycommand/ADR-001
*/

// Import from lease-types, NOT meta/coordinator: the latter evaluates
// `cloudflare:workers`, which does not resolve on Node and would crash the
// daemon at module load, before main() and before any log line.
import { META_LEADER_ROLE, type StoredLease } from '../meta/lease-types';

export { META_LEADER_ROLE };

export const POLICY_BLOCKED_COORDINATOR_UNAVAILABLE =
'POLICY_BLOCKED_COORDINATOR_UNAVAILABLE';

export interface CoordinatorEnv {
/** Base URL of the ChittyCommand worker, e.g. https://command.chitty.cc */
COORDINATOR_URL?: string;
/** Bearer token for the coordinator routes. Broker-provided; never inlined. */
COORDINATOR_TOKEN?: string;
}

/** Identical to NodeLease in daemon/leader.ts. */
export interface NodeLease {
role: string;
nodeId: string;
nodeDescriptor: string | null;
sessionId: string | null;
claimedAt: Date;
heartbeatAt: Date;
leaseExpiresAt: Date;
metadata: Record<string, unknown>;
}

export interface ClaimOptions {
nodeId: string;
nodeDescriptor?: string;
sessionId?: string;
leaseSeconds?: number;
role?: string;
metadata?: Record<string, unknown>;
}

/**
* Both the URL and the token are required. A configured URL with no token
* yields a 401 on every call, which loop.ts logs as a claim error and retries
* forever — a permanently dead daemon whose logs read like a transient auth
* blip. Fail closed on the config error instead, with a distinguishable code.
*/
function requireConfig(env: CoordinatorEnv): { url: string; token: string } {
const url = env.COORDINATOR_URL?.replace(/\/+$/, '');
if (!url) throw new Error(POLICY_BLOCKED_COORDINATOR_UNAVAILABLE);
if (!env.COORDINATOR_TOKEN) throw new Error(POLICY_BLOCKED_COORDINATOR_UNAVAILABLE);
return { url, token: env.COORDINATOR_TOKEN };
}

async function call<T>(
env: CoordinatorEnv,
method: 'GET' | 'POST',
path: string,
body?: unknown,
): Promise<T> {
const { url, token } = requireConfig(env);
const headers: Record<string, string> = {
'content-type': 'application/json',
authorization: `Bearer ${token}`,
};

const res = await fetch(`${url}/api/meta/coordinator${path}`, {
method,
headers,
...(body === undefined ? {} : { body: JSON.stringify(body) }),
});

if (!res.ok) {
throw new Error(
`[daemon/coordinator-lease] ${method} ${path} failed: ${res.status} ${await res.text()}`,
);
}
return (await res.json()) as T;
}

/** Rehydrate ISO strings into Dates. Returns null for an unheld lease. */
function toLease(stored: StoredLease | null): NodeLease | null {
if (!stored?.nodeId || !stored.claimedAt || !stored.heartbeatAt || !stored.leaseExpiresAt) {
return null;
}
return {
role: stored.role,
nodeId: stored.nodeId,
nodeDescriptor: stored.nodeDescriptor,
sessionId: stored.sessionId,
claimedAt: new Date(stored.claimedAt),
heartbeatAt: new Date(stored.heartbeatAt),
leaseExpiresAt: new Date(stored.leaseExpiresAt),
metadata: stored.metadata ?? {},
};
}

export async function claimLeadership(
env: CoordinatorEnv,
options: ClaimOptions,
): Promise<NodeLease | null> {
if (!options?.nodeId) throw new Error('[daemon/coordinator-lease] nodeId is required');
return toLease(
await call<StoredLease | null>(env, 'POST', '/claim', {
nodeId: options.nodeId,
nodeDescriptor: options.nodeDescriptor ?? null,
sessionId: options.sessionId ?? null,
leaseSeconds: options.leaseSeconds,
role: options.role,
metadata: options.metadata ?? {},
}),
);
}

export async function heartbeat(
env: CoordinatorEnv,
nodeId: string,
options: { role?: string; leaseSeconds?: number; sessionId?: string | null } = {},
): Promise<NodeLease | null> {
if (!nodeId) throw new Error('[daemon/coordinator-lease] nodeId is required for heartbeat');
return toLease(
await call<StoredLease | null>(env, 'POST', '/heartbeat', {
nodeId,
role: options.role,
leaseSeconds: options.leaseSeconds,
sessionId: options.sessionId ?? null,
}),
);
}

export async function releaseLeadership(
env: CoordinatorEnv,
nodeId: string,
options: { role?: string; sessionId?: string | null } = {},
): Promise<boolean> {
if (!nodeId) throw new Error('[daemon/coordinator-lease] nodeId is required for release');
const res = await call<{ released: boolean }>(env, 'POST', '/release', {
nodeId,
role: options.role,
sessionId: options.sessionId ?? null,
});
return res.released === true;
}

export async function describeLease(
env: CoordinatorEnv,
options: { role?: string } = {},
): Promise<NodeLease | null> {
const role = options.role ?? META_LEADER_ROLE;
return toLease(
await call<StoredLease | null>(
env,
'GET',
`/describe?role=${encodeURIComponent(role)}`,
),
);
}
221 changes: 221 additions & 0 deletions meta/coordinator.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,221 @@
/**
* CommandCoordinator — Durable Object arbiter for meta-orchestrator role leases.
*
* Replaces the Neon `cc_node_leases` table (daemon/leader.ts) as the *arbiter*
* of who holds a role. A Durable Object is already a strongly-consistent,
* single-threaded singleton, so the properties the Neon lease bought us come
* free:
*
* - mutual exclusion: the DO serializes requests; no two claimers race
* - durability: DO storage survives eviction and restart
* - availability: runs on the Cloudflare edge, not on operator hardware
*
* ADR-001 elected a leader across `chittymini-01..06` via Neon leases so the
* coordinator would never be down. That mechanism is unnecessary here — there
* is no fleet to elect across, because the DO *is* the leader. Nodes remain
* meaningful as executors that pull work (they have local filesystem and repo
* access, which is the real reason to run on hardware); they are no longer
* candidates for leadership.
*
* Wire semantics are a deliberate 1:1 port of the SQL in daemon/leader.ts,
* including two behaviours established by prior review:
* - heartbeat requires (role, nodeId, sessionId) to match the holder
* (codex-p2 PR#101 finding-5)
* - release requires the same triple (codex-p2 PR#101 finding-2)
*
* @canonical-uri chittycanon://docs/architecture/chittycommand/ADR-001
* @canon chittycanon://gov/governance#core-types — a node is a Location (L);
* a lease claim is an Event (E).
*/

import { DurableObject } from 'cloudflare:workers';

import {
DEFAULT_LEASE_SECONDS,
MAX_LEASE_SECONDS,
META_LEADER_ROLE,
MIN_LEASE_SECONDS,
normalizeLeaseSeconds,
type ClaimBody,
type StoredLease,
} from './lease-types';

// Re-exported so existing importers of this module keep working. The
// definitions live in lease-types.ts because the daemon runs on Node and
// cannot evaluate `cloudflare:workers`.
export {
DEFAULT_LEASE_SECONDS,
MAX_LEASE_SECONDS,
META_LEADER_ROLE,
MIN_LEASE_SECONDS,
normalizeLeaseSeconds,
};
export type { ClaimBody, StoredLease };

/** Storage key for a role's lease. */
const keyFor = (role: string) => `lease:${role}`;

export class CommandCoordinator extends DurableObject {
async fetch(request: Request): Promise<Response> {
const url = new URL(request.url);
// Anchor on the FIRST '/coordinator' segment. A greedy match here let
// `/api/meta/coordinator/a/coordinator/release` dispatch `release`.
const marker = '/coordinator';
const at = url.pathname.indexOf(marker);
const path = at === -1 ? '/' : url.pathname.slice(at + marker.length) || '/';

try {
switch (`${request.method} ${path}`) {
case 'POST /claim':
return json(await this.claim(await readJson<ClaimBody>(request)));
case 'POST /heartbeat':
return json(await this.heartbeat(await readJson(request)));
case 'POST /release':
return json({ released: await this.release(await readJson(request)) });
case 'GET /describe':
return json(await this.describe(url.searchParams.get('role') ?? META_LEADER_ROLE));
default:
return json({ error: 'not_found', path }, 404);
}
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
return json({ error: 'coordinator_error', message }, 400);
}
}

private async read(role: string): Promise<StoredLease | undefined> {
return this.ctx.storage.get<StoredLease>(keyFor(role));
}

/**
* Claim `role`. Succeeds when the role is unheld, already held by this same
* node, or the incumbent's lease has expired.
*
* Parity note: `claimedAt` is preserved across takeover, mirroring
* `COALESCE(claimed_at, NOW())` in the SQL. That means a takeover from an
* expired holder reports the *previous* holder's claim time. This is
* deliberate and is not a defect: `claimedAt` reads as "when this role was
* first continuously held", which is a coherent semantic.
*
* It is also load-bearing, not diagnostic — daemon/coordinator-lease.ts
* returns null when `claimedAt` is falsy, so it participates in the
* lease/no-lease decision. Any future change has a second consumer.
*/
async claim(body: ClaimBody): Promise<StoredLease | null> {
if (!body?.nodeId) throw new Error('[meta/coordinator] nodeId is required');

const role = body.role ?? META_LEADER_ROLE;
const leaseSeconds = normalizeLeaseSeconds(body.leaseSeconds);
const now = Date.now();
const current = await this.read(role);

// A corrupted timestamp parses to NaN, and every NaN comparison is false —
// which would leave the role permanently unclaimable. Postgres typed this
// column so the SQL had no such failure mode; treat unparseable as expired.
const expiresAt = current?.leaseExpiresAt ? Date.parse(current.leaseExpiresAt) : NaN;
const expired = !Number.isFinite(expiresAt) || expiresAt < now;
const claimable = !current?.nodeId || current.nodeId === body.nodeId || expired;
if (!claimable) return null;

const nowIso = new Date(now).toISOString();
const lease: StoredLease = {
role,
nodeId: body.nodeId,
nodeDescriptor: body.nodeDescriptor ?? null,
sessionId: body.sessionId ?? null,
claimedAt: current?.claimedAt ?? nowIso,
heartbeatAt: nowIso,
leaseExpiresAt: new Date(now + leaseSeconds * 1000).toISOString(),
metadata: body.metadata ?? {},
};

await this.ctx.storage.put(keyFor(role), lease);
return lease;
}

/**
* Extend the lease. Returns null when this node is no longer the holder, or
* when `sessionId` does not match the session recorded on the lease — a
* restarted process reusing a nodeId cannot heartbeat over a fresh leader.
*/
async heartbeat(body: {
nodeId: string;
role?: string;
leaseSeconds?: number;
sessionId?: string | null;
}): Promise<StoredLease | null> {
if (!body?.nodeId) throw new Error('[meta/coordinator] nodeId is required for heartbeat');

const role = body.role ?? META_LEADER_ROLE;
const current = await this.read(role);
if (!current || current.nodeId !== body.nodeId) return null;
if (current.sessionId !== (body.sessionId ?? null)) return null;

const now = Date.now();
const leaseSeconds = normalizeLeaseSeconds(body.leaseSeconds);
const lease: StoredLease = {
...current,
heartbeatAt: new Date(now).toISOString(),
leaseExpiresAt: new Date(now + leaseSeconds * 1000).toISOString(),
};

await this.ctx.storage.put(keyFor(role), lease);
return lease;
}

/**
* Release the role. Only the holding (nodeId, sessionId) pair may release;
* a different node or a newer session of the same node is a no-op.
*/
async release(body: {
nodeId: string;
role?: string;
sessionId?: string | null;
}): Promise<boolean> {
if (!body?.nodeId) throw new Error('[meta/coordinator] nodeId is required for release');

const role = body.role ?? META_LEADER_ROLE;
const current = await this.read(role);
if (!current || current.nodeId !== body.nodeId) return false;
if (current.sessionId !== (body.sessionId ?? null)) return false;

await this.ctx.storage.put(keyFor(role), {
role,
nodeId: null,
nodeDescriptor: null,
sessionId: null,
claimedAt: null,
heartbeatAt: null,
leaseExpiresAt: null,
metadata: current.metadata,
} satisfies StoredLease);
return true;
}

/**
* Inspect the lease without mutating it. Returns null when unheld.
*
* Parity note: an *expired but unreleased* lease is still returned, matching
* `describeLease()` in daemon/leader.ts, which filters only on `node_id`.
* Callers must not treat a non-null result as proof of live leadership.
*/
async describe(role: string = META_LEADER_ROLE): Promise<StoredLease | null> {
const current = await this.read(role);
return current?.nodeId ? current : null;
}
}

async function readJson<T>(request: Request): Promise<T> {
try {
return (await request.json()) as T;
} catch {
throw new Error('invalid JSON body');
}
}

function json(data: unknown, status = 200): Response {
return new Response(JSON.stringify(data), {
status,
headers: { 'content-type': 'application/json' },
});
}
Loading
Loading