feat(control-plane): cron-triggered wake/poll orchestration for hosted AMS tenants

control-plane/src/ams-wake.ts

@@ -0,0 +1,120 @@
+// Cron-triggered wake orchestration for hosted AMS tenants (#7182, the control-plane half -- the miner-side
+// hosted entry point it wakes, packages/loopover-miner/bin/loopover-miner-hosted.ts, is a separate, already-
+// shipped PR). Cloudflare Cron Triggers fire ONE global `scheduled()` handler on a fixed schedule (no
+// per-resource cron primitive exists) -- so per-tenant cadence lives as DATA on each AMS tenant's own
+// `amsSchedule` (tenant-registry.ts), and this module's job every tick is: find whichever tenants are
+// currently due, wake each one's container with the right one-shot command, wait for it to finish, and
+// record what happened (#7182's own "0=success/2=failure" exit-code alerting contract, unmodified).
+//
+// Endpoint/state semantics below follow @cloudflare/containers' documented Container API (start/getState) at
+// the time this was written -- verify against a live account before the first real deploy (mirrors
+// neon-database-driver.ts's identical header-comment caveat); every test here mocks this boundary.
+import type { Product } from "./tenant-provisioning-driver.js";
+import type { TenantRegistry, TenantRegistryRecord } from "./tenant-registry.js";
+
+/** The slice of a real Container DO's RPC surface this module actually calls -- a SEPARATE small local
+ *  interface from container-driver.ts's own `ContainerStubLike` (that one never needs `getState()`; this one
+ *  needs nothing else). Mirrors this package's established "local interface, no SDK import" convention. */
+export type WakeStubLike = {
+  start(options?: { entrypoint?: string[] }): Promise<void>;
+  getState(): Promise<{ status: string; exitCode?: number }>;
+};
+
+export type WakeNamespaceLike = {
+  getByName(name: string): WakeStubLike;
+};
+
+export type AmsWakeConfig = {
+  binding: WakeNamespaceLike;
+  registry: TenantRegistry;
+  /** Overridable for tests only -- production always uses real wall-clock time and real delays. */
+  pollIntervalMs?: number;
+  pollTimeoutMs?: number;
+  now?: () => Date;
+};
+
+export type AmsWakeResult = {
+  tenant: TenantRegistryRecord["tenant"];
+  ranAt: string;
+  /** The hosted entry point's own exit code, or `undefined` if the container never reached a stopped state
+   *  before `pollTimeoutMs` elapsed (a real failure mode in its own right -- surfaced as `timedOut`, not
+   *  silently coerced into a fake exit code). */
+  exitCode: number | undefined;
+  timedOut: boolean;
+};
+
+const HOSTED_ENTRY_BIN = "loopover-miner-hosted";
+const DEFAULT_POLL_INTERVAL_MS = 1_000;
+// A generous ceiling for a real discover/manage-poll/attempt cycle -- long enough that a real, working run
+// almost never hits it, short enough that a genuinely hung container doesn't block this tick's remaining
+// tenants indefinitely (this loop processes due tenants one at a time, not in parallel; see wakeDueAmsTenants).
+const DEFAULT_POLL_TIMEOUT_MS = 10 * 60 * 1000;
+
+/** Same `${product}:${name}` composite container-driver.ts's own `instanceNameFor` derives -- duplicated
+ *  (not imported) because container-driver.ts's version takes a `TenantProvisioningRequest`, a shape this
+ *  module has no reason to construct just to call it. */
+function instanceNameFor(name: string, product: Product): string {
+  return `${product}:${name}`;
+}
+
+/** A tenant is due when it's an active AMS tenant with a schedule whose `nextDueAt` has arrived. Anything
+ *  else (a different product, a torn-down/provisioning tenant, no schedule at all, or a schedule that isn't
+ *  due yet) is silently skipped -- this is a routine filter, not an error condition. */
+function isDue(record: TenantRegistryRecord, now: Date): boolean {
+  return record.product === "ams" && record.state === "active" && record.amsSchedule !== undefined && new Date(record.amsSchedule.nextDueAt).getTime() <= now.getTime();
+}
+
+/** Polls `getState()` until the container reaches a stopped state (with or without an exit code -- either
+ *  means the one-shot process is done running) or `timeoutMs` elapses, whichever comes first. Returns
+ *  `timedOut: true` only when the deadline was actually hit -- a genuinely-finished container reporting no
+ *  exit code (a bare `"stopped"` status) is a different, non-timeout outcome, even though both cases leave
+ *  `exitCode` as `undefined`. */
+async function pollForExitCode(stub: WakeStubLike, pollIntervalMs: number, timeoutMs: number): Promise<{ exitCode: number | undefined; timedOut: boolean }> {
+  const deadline = Date.now() + timeoutMs;
+  for (;;) {
+    const state = await stub.getState();
+    if (state.status === "stopped" || state.status === "stopped_with_code") return { exitCode: state.exitCode, timedOut: false };
+    if (Date.now() >= deadline) return { exitCode: undefined, timedOut: true };
+    await new Promise((resolve) => setTimeout(resolve, pollIntervalMs));
+  }
+}
+
+/** Wakes every currently-due AMS tenant, one at a time (deliberately sequential, not `Promise.all` -- a
+ *  single Cloudflare Cron Trigger invocation has a bounded wall-clock budget shared across every tenant this
+ *  tick processes; running them concurrently would trade a slow tick for cross-tenant resource contention on
+ *  shared infra this module has no visibility into). Advances each woken tenant's `nextDueAt` from the tick's
+ *  OWN start time (not the run's completion time) so schedule drift doesn't accumulate when a cycle runs long.
+ */
+export async function wakeDueAmsTenants(config: AmsWakeConfig): Promise<AmsWakeResult[]> {
+  const now = config.now ?? (() => new Date());
+  const pollIntervalMs = config.pollIntervalMs ?? DEFAULT_POLL_INTERVAL_MS;
+  const pollTimeoutMs = config.pollTimeoutMs ?? DEFAULT_POLL_TIMEOUT_MS;
+  const tickStartedAt = now();
+
+  const records = await config.registry.list();
+  const results: AmsWakeResult[] = [];
+
+  for (const record of records) {
+    if (!isDue(record, tickStartedAt)) continue;
+    const schedule = record.amsSchedule!;
+
+    const stub = config.binding.getByName(instanceNameFor(record.tenant.name, record.product));
+    await stub.start({ entrypoint: [HOSTED_ENTRY_BIN, schedule.command, ...schedule.args] });
+    const { exitCode, timedOut } = await pollForExitCode(stub, pollIntervalMs, pollTimeoutMs);
+
+    const ranAt = now().toISOString();
+    await config.registry.upsert({
+      ...record,
+      amsSchedule: {
+        ...schedule,
+        lastRunAt: ranAt,
+        lastExitCode: exitCode,
+        nextDueAt: new Date(tickStartedAt.getTime() + schedule.intervalMs).toISOString(),
+      },
+      updatedAt: ranAt,
+    });
+    results.push({ tenant: record.tenant, ranAt, exitCode, timedOut });
+  }
+
+  return results;
+}

control-plane/src/container-driver.ts

@@ -48,7 +48,10 @@ export type ContainerDriver = {
   containerExists(request: TenantProvisioningRequest): Promise<boolean>;
 };
 
-function instanceNameFor(request: TenantProvisioningRequest): string {
+/** The `${product}:${name}` composite key a tenant's Container DO is addressed by -- exported so other
+ *  modules that need to reach the SAME instance (e.g. ams-wake.ts's cron-triggered wake) derive it identically
+ *  rather than duplicating the format and risking drift. Matches tenant-registry.ts's own `instanceKeyFor`. */
+export function instanceNameFor(request: TenantProvisioningRequest): string {
   return `${request.product}:${request.tenant.name}`;
 }
 

control-plane/src/http-app.ts

@@ -8,9 +8,14 @@
 //
 // Deliberately never echoes a tenant's database connection details (host/user/password/connectionString) in
 // any response: `provisionTenant`'s result carries them (#7653) so a caller doesn't lose them, but this admin
-// HTTP surface only returns the safe `{tenant, product, state}` triple. Properly storing/distributing those
-// credentials is #7852's job (the generalized secret broker) -- until it lands, this transport intentionally
-// does not create a new place for them to leak.
+// HTTP surface only returns the safe `{tenant, product, state}` triple (plus `amsSchedule` when set, #7182 --
+// a cron cadence and command name, never a secret). Properly storing/distributing credentials is #7852's job
+// (the generalized secret broker) -- until it lands, this transport intentionally does not create a new
+// place for them to leak.
+//
+// `POST /v1/tenants` also accepts an optional `schedule` field (#7182), valid only for `product: "ams"`:
+// configures the new tenant's cron-wake cadence at creation time. ams-wake.ts's `scheduled()`-triggered
+// handler is what actually reads and acts on it later -- this route only validates and stores it.
 import { Hono } from "hono";
 import { normalizeSharedSecret, verifyBearer } from "./auth.js";
 import {
@@ -19,7 +24,7 @@ import {
   type ProvisioningPagerDutyOptions,
 } from "./provisioning.js";
 import type { Product, TenantProvisioningDriver } from "./tenant-provisioning-driver.js";
-import type { TenantRegistry, TenantRegistryRecord } from "./tenant-registry.js";
+import type { AmsCycleSchedule, TenantRegistry, TenantRegistryRecord } from "./tenant-registry.js";
 
 export type TenantHttpAppDeps = {
   driver: TenantProvisioningDriver;
@@ -31,8 +36,36 @@ export type TenantHttpAppDeps = {
   pagerDuty?: ProvisioningPagerDutyOptions;
 };
 
-function safeRecord(record: Pick<TenantRegistryRecord, "tenant" | "product" | "state">): Record<string, unknown> {
-  return { tenant: record.tenant, product: record.product, state: record.state };
+function safeRecord(record: Pick<TenantRegistryRecord, "tenant" | "product" | "state" | "amsSchedule">): Record<string, unknown> {
+  return { tenant: record.tenant, product: record.product, state: record.state, ...(record.amsSchedule ? { amsSchedule: record.amsSchedule } : {}) };
+}
+
+/** The only `command` names #7182's hosted entry point (loopover-miner-hosted) actually dispatches --
+ *  mirrors packages/loopover-miner/lib/hosted-entry.ts's own `HOSTED_CYCLE_COMMANDS` keys exactly. Kept as a
+ *  plain string literal here (not imported) since this package has no cross-package type/value coupling with
+ *  loopover-miner anywhere else -- a drift between the two lists would only matter if someone edits one
+ *  without the other, which is why both list comments point at each other. */
+const HOSTED_CYCLE_COMMANDS = ["discover", "manage-poll", "attempt"] as const;
+
+/** Validated body of `POST /v1/tenants`'s optional `schedule` field (#7182): configures a NEW AMS tenant's
+ *  cron-wake cadence at creation time. `undefined` input (the field omitted entirely) is valid and means "no
+ *  schedule yet" -- an AMS tenant with no schedule simply never gets woken, which is a legitimate state, not
+ *  an error. `intervalMs` has no configured maximum: an operator setting an absurdly long interval is their
+ *  own call to make, not something this validation second-guesses. */
+function parseScheduleRequest(value: unknown): AmsCycleSchedule | string | undefined {
+  if (value === undefined) return undefined;
+  if (value === null || typeof value !== "object" || Array.isArray(value)) return "schedule must be a JSON object";
+  const { command, args, intervalMs } = value as Record<string, unknown>;
+  if (typeof command !== "string" || !(HOSTED_CYCLE_COMMANDS as readonly string[]).includes(command)) {
+    return `schedule.command must be one of: ${HOSTED_CYCLE_COMMANDS.join(", ")}`;
+  }
+  if (args !== undefined && (!Array.isArray(args) || !args.every((value): value is string => typeof value === "string"))) {
+    return "schedule.args must be an array of strings";
+  }
+  if (typeof intervalMs !== "number" || !Number.isFinite(intervalMs) || intervalMs <= 0) {
+    return "schedule.intervalMs must be a positive number of milliseconds";
+  }
+  return { command, args: Array.isArray(args) ? args : [], intervalMs, nextDueAt: new Date().toISOString() };
 }
 
 /** Validated body of `POST /v1/tenants/rollout` (#4898): an explicit tenant-name list (no percentage/canary
@@ -81,9 +114,12 @@ export function createTenantHttpApp(deps: TenantHttpAppDeps): Hono {
   app.post("/v1/tenants", async (c) => {
     const body: unknown = await c.req.json().catch(() => null);
     if (body === null || typeof body !== "object") return c.json({ error: "invalid_json" }, 400);
-    const { name, product } = body as Record<string, unknown>;
+    const { name, product, schedule: scheduleInput } = body as Record<string, unknown>;
     if (typeof name !== "string" || !name.trim()) return c.json({ error: "invalid_request", message: "name is required" }, 400);
     if (typeof product !== "string" || !product.trim()) return c.json({ error: "invalid_request", message: "product is required" }, 400);
+    const schedule = parseScheduleRequest(scheduleInput);
+    if (typeof schedule === "string") return c.json({ error: "invalid_request", message: schedule }, 400);
+    if (schedule && product !== "ams") return c.json({ error: "invalid_request", message: 'schedule is only valid for product "ams"' }, 400);
 
     // Not idempotent by design (tenant-client.ts's own doc comment: "a create is not idempotent, so it must
     // not be silently re-sent") -- a currently-active tenant of the same name *and product* is a real conflict,
@@ -94,8 +130,9 @@ export function createTenantHttpApp(deps: TenantHttpAppDeps): Hono {
 
     const result = await provisionTenant({ name }, product, deps.driver, deps.pagerDuty ?? {});
     const now = new Date().toISOString();
-    await deps.registry.upsert({ tenant: result.tenant, product: result.product, state: result.state, createdAt: now, updatedAt: now });
-    return c.json(safeRecord(result), 201);
+    const record: TenantRegistryRecord = { tenant: result.tenant, product: result.product, state: result.state, createdAt: now, updatedAt: now, ...(schedule ? { amsSchedule: schedule } : {}) };
+    await deps.registry.upsert(record);
+    return c.json(safeRecord(record), 201);
   });
 
   app.get("/v1/tenants", async (c) => {

control-plane/src/index.ts

@@ -55,6 +55,7 @@ export {
   createContainerDriver,
   createTenantContainer,
   destroyTenantContainer,
+  instanceNameFor,
   PINNED_VERSION_ENV_VAR,
   tenantContainerExists,
   type ContainerDriver,
@@ -65,9 +66,17 @@ export {
 export {
   createFakeTenantRegistry,
   createKvTenantRegistry,
+  type AmsCycleSchedule,
   type KvNamespaceLike,
   type TenantRegistry,
   type TenantRegistryRecord,
 } from "./tenant-registry.js";
 export { createTenantHttpApp, type TenantHttpAppDeps } from "./http-app.js";
 export { normalizeSharedSecret, verifyBearer } from "./auth.js";
+export {
+  wakeDueAmsTenants,
+  type AmsWakeConfig,
+  type AmsWakeResult,
+  type WakeNamespaceLike,
+  type WakeStubLike,
+} from "./ams-wake.js";

control-plane/src/tenant-registry.ts

@@ -6,12 +6,33 @@
 // credential store (that's #7852's job, via the generalized broker).
 import type { Product, Tenant, TenantLifecycleState } from "./tenant-provisioning-driver.js";
 
+/** One AMS tenant's cron-wake configuration (#7182) -- ORB tenants never have this (they're woken by
+ *  incoming webhooks, #7181, not a schedule). `command`/`args` are forwarded verbatim to
+ *  `loopover-miner-hosted` (packages/loopover-miner/bin/loopover-miner-hosted.ts) as its own argv -- this
+ *  package deliberately does not import loopover-miner's `HostedCycleCommand` type (no cross-package type
+ *  coupling in this codebase's existing convention), so `command` is validated as a plain string against the
+ *  same three known names at the HTTP layer instead (see http-app.ts). */
+export type AmsCycleSchedule = {
+  command: string;
+  args: string[];
+  intervalMs: number;
+  /** When this tenant is next due to be woken. Advances by `intervalMs` after every run (#7182's own
+   *  "wake, run one cycle, sleep" model), regardless of whether that run succeeded. */
+  nextDueAt: string;
+  lastRunAt?: string;
+  /** The hosted entry point's own exit code from the most recent run (0=success, 2=failure, per
+   *  `docs/unattended-scheduling.md`'s existing contract) -- `undefined` until the first run, or if the most
+   *  recent run timed out waiting for the container to stop. */
+  lastExitCode?: number;
+};
+
 export type TenantRegistryRecord = {
   tenant: Tenant;
   product: Product;
   state: TenantLifecycleState;
   createdAt: string;
   updatedAt: string;
+  amsSchedule?: AmsCycleSchedule;
 };
 
 export interface TenantRegistry {

control-plane/src/worker.ts

@@ -7,6 +7,7 @@
 // Not unit-tested: exercised only by real Cloudflare Workers/KV/Containers infrastructure, matching
 // packages/discovery-index/src/worker.ts's own identical exclusion (see scripts/control-plane-coverage.mjs).
 import { Container } from "@cloudflare/containers";
+import { wakeDueAmsTenants } from "./ams-wake.js";
 import { createTenantProvisioningDriver } from "./driver-factory.js";
 import { createTenantHttpApp } from "./http-app.js";
 import { createKvTenantRegistry } from "./tenant-registry.js";
@@ -67,4 +68,18 @@ export default {
     });
     return app.fetch(request, env);
   },
+
+  // Cron Trigger handler (#7182, wrangler.jsonc's `triggers.crons`): one global tick, on the schedule
+  // wrangler.jsonc declares, checks every AMS tenant's own `amsSchedule.nextDueAt` (there is no per-resource
+  // Cron Trigger primitive to register one per tenant) and wakes whichever ones are due. `ctx.waitUntil`
+  // keeps the tick alive until every due tenant's cycle finishes, since `wakeDueAmsTenants` itself awaits
+  // each one before the Worker would otherwise be allowed to shut the invocation down.
+  async scheduled(_event: ScheduledEvent, env: Env, ctx: ExecutionContext): Promise<void> {
+    ctx.waitUntil(
+      wakeDueAmsTenants({
+        binding: env.AMS_TENANT_CONTAINER,
+        registry: createKvTenantRegistry(env.TENANT_REGISTRY),
+      }),
+    );
+  },
 };