feat(node): connection observability — last-seen per ticket, gated-connection status
- commit
- 0f11ec5c96182e60be936a1b06418beca3f4874d
- parent
- c17df3e58a06c14e0d81fdf82015ab44b6b54e62
- parent
- 09e5c098ccef383cc99d6ee906a87f92e229c25a
- author
- FelineStateMachine <dami.etoile@gmail.com>
| M | cli.ts | 9 | +++- |
| M | docs/ | 13 | +++- |
| M | src/ | 26 | ++++++-- |
| M | src/ | 16 | +++++ |
| A | src/ | 87 | ++++++++++++++++++++++++++++ |
| A | tests/ | 93 | ++++++++++++++++++++++++++++++ |
| M | tests/ | 7 | ++- |
@@ -32,6 +32,7 @@ import {
3232 isRevokedByLineage,3333 looksLikeAppTicket,3434 } from "./src/appticket.ts";35+import { readTicketActivity } from "./src/ticket-activity.ts";3536 3637 const USAGE = `lofi-node — self-hostable sync node for lofi apps3738 @@ -307,13 +308,19 @@ async function cmdTicket(args: Args) {
307308 console.log("no tickets issued");308309 return;309310 }311+ // Last-seen comes from the daemon-owned sidecar; absent (never seen, or312+ // the node has not flushed yet) prints as "-".313+ const activity = args.dir ? await readTicketActivity(args.dir) : new Map<string, string>();310314 for (const t of tickets) {311315 // Lineage-aware: a derived ticket shows REVOKED once its parent is.312316 const dead = isRevokedByLineage(t, tickets);317+ const lastSeen = activity.get(t.id) ?? "-";313318 console.log(314319 `${t.id} ${dead ? "REVOKED" : "active "} ${315320 (t.scope ?? "sync").padEnd(9)316- } ${t.createdAt} ${t.label ?? ""}${t.parentId ? ` [from ${t.parentId}]` : ""}`,321+ } ${t.createdAt} seen ${lastSeen} ${t.label ?? ""}${322+ t.parentId ? ` [from ${t.parentId}]` : ""323+ }`,317324 );318325 }319326 return;@@ -54,9 +54,12 @@ message). Re-electable at runtime; the public port does not change.
5454 `--provision` mints a provision-scoped ticket that can also administer the store5555 ([scopes](tickets-explained.md)).5656 57-`list` shows every issued ticket: id, scope, label, issued time, and revocation state — including58-tickets revoked through their lineage, since a [derived sync ticket](tickets-explained.md) dies with59-its parent provision ticket. `revoke <id>` invalidates one (and everything derived from it); a57+`list` shows every issued ticket: id, scope, label, issued time, last-seen time, and revocation58+state — including tickets revoked through their lineage, since a59+[derived sync ticket](tickets-explained.md) dies with its parent provision ticket. Last-seen comes60+from a daemon-written sidecar (`ticket-activity.json`, flushed about once a minute), so it answers61+"is that phone still using this ticket?" without any daemon IPC; `-` means the ticket has not been62+seen since the sidecar existed. `revoke <id>` invalidates one (and everything derived from it); a6063 running daemon picks the change up within seconds and closes the affected live connections with6164 close code 4001 — no restart, no IPC.6265 @@ -64,4 +67,6 @@ close code 4001 — no restart, no IPC.
6467 6568 Prints configuration and live state without changing anything: access mode, storage, relay election,6669 issued tickets, upstream election, and mesh state (up with connection stats, off, or unavailable67-with the reason).70+with the reason). Embedders get the same data programmatically from `node.status()`, which also71+carries `gatedConnections` — live gated connection counts and last-seen per ticket, joined with72+labels.@@ -67,13 +67,16 @@ export interface GateOptions {
6767 /** The node's iroh EndpointTicket, carried on derived tickets exactly like6868 * node-issued ones. */6969 nodeTicket?: string;70+ /** Activity sink notified on every authenticated request and WS open —71+ * feeds the daemon-owned last-seen sidecar. */72+ activity?: { note(ticketId: string): void };7073 }7174 7275 export interface Gate {7376 port: number;7477 url: string;75- /** Live gated WS connections per ticket id. */76- stats(): { ticketId: string; connections: number }[];78+ /** Live gated WS connections and last observed activity per ticket id. */79+ stats(): { ticketId: string; connections: number; lastSeenAt?: string }[];7780 close(): Promise<void>;7881 }7982 @@ -320,11 +323,18 @@ export function startGate(options: GateOptions): Gate {
320323 return token;321324 };322325 326+ const lastSeenAt = new Map<string, string>();327+ const seen = (ticketId: string) => {328+ lastSeenAt.set(ticketId, new Date().toISOString());329+ options.activity?.note(ticketId);330+ };331+323332 const track = (ticketId: string, socket: WebSocket) => {324333 let set = liveByTicket.get(ticketId);325334 if (!set) liveByTicket.set(ticketId, set = new Set());326335 set.add(socket);327336 socket.addEventListener("close", () => {337+ seen(ticketId);328338 set.delete(socket);329339 if (set.size === 0) liveByTicket.delete(ticketId);330340 });@@ -399,6 +409,7 @@ export function startGate(options: GateOptions): Gate {
399409 return unauthorized();400410 }401411 const scope = verdict.record.scope ?? "sync";412+ seen(verdict.record.id);402413 let strippedPath = rest || "/";403414 // Possession-bound tickets: the bare secret opens only the404415 // proof-of-possession exchange; everything else must ride a connect@@ -567,11 +578,14 @@ export function startGate(options: GateOptions): Gate {
567578 return {568579 port,569580 url: `http://127.0.0.1:${port}`,570- stats: () =>571- [...liveByTicket.entries()].map(([ticketId, sockets]) => ({581+ stats: () => {582+ const ids = new Set([...liveByTicket.keys(), ...lastSeenAt.keys()]);583+ return [...ids].map((ticketId) => ({572584 ticketId,573- connections: sockets.size,574- })),585+ connections: liveByTicket.get(ticketId)?.size ?? 0,586+ ...(lastSeenAt.has(ticketId) ? { lastSeenAt: lastSeenAt.get(ticketId) } : {}),587+ }));588+ },575589 close: async () => {576590 clearInterval(sweep);577591 await server.shutdown();@@ -21,6 +21,7 @@ import {
2121 validateStorage,2222 } from "./config.ts";2323 import { type Gate, startGate } from "./gate.ts";24+import { TicketActivity } from "./ticket-activity.ts";2425 import {2526 type AppTicketRecord,2627 AppTicketStore,@@ -93,6 +94,14 @@ export interface SyncNodeStatus {
9394 access: "open" | "ticket";9495 jazz: { url: string; port: number; storage: StorageConfig };9596 tickets: AppTicketInfo[];97+ /** Live gated connections and last observed activity, joined with labels;98+ * empty on open-mode nodes. */99+ gatedConnections: {100+ ticketId: string;101+ connections: number;102+ lastSeenAt?: string;103+ label?: string;104+ }[];96105 upstream: UpstreamConfig;97106 mesh: MeshStatus;98107 }@@ -259,6 +268,7 @@ export async function createSyncNode(options: SyncNodeOptions): Promise<SyncNode
259268 260269 // 3b. App tickets + gate (ticket mode).261270 const ticketStore = await AppTicketStore.load(options.dataDir);271+ const ticketActivity = await TicketActivity.load(options.dataDir);262272 let gate: Gate | null = null;263273 if (gated) {264274 gate = startGate({@@ -270,6 +280,7 @@ export async function createSyncNode(options: SyncNodeOptions): Promise<SyncNode
270280 adminSecret: options.adminSecret,271281 publicBase: options.publicUrl,272282 nodeTicket: ticketString ?? undefined,283+ activity: ticketActivity,273284 });274285 }275286 @@ -362,6 +373,10 @@ export async function createSyncNode(options: SyncNodeOptions): Promise<SyncNode
362373 access,363374 jazz: { url: jazz.url, port: jazz.port, storage },364375 tickets: ticketSnapshot,376+ gatedConnections: (gate?.stats() ?? []).map((entry) => ({377+ ...entry,378+ label: ticketSnapshot.find((t) => t.id === entry.ticketId)?.label,379+ })),365380 upstream,366381 mesh: mesh.state === "up"367382 ? {@@ -375,6 +390,7 @@ export async function createSyncNode(options: SyncNodeOptions): Promise<SyncNode
375390 acceptor?.close();376391 await tunnelListener?.close();377392 await gate?.close();393+ await ticketActivity.flush().catch(() => {});378394 await jazz.stop();379395 await irohNode?.close();380396 })()),@@ -0,0 +1,87 @@
1+// Last-seen activity per ticket, persisted in a daemon-owned sidecar so the2+// CLI can answer "is that phone still using this ticket?" without IPC. A3+// separate file — never tickets.json — because tickets.json is CLI-written4+// and mtime-hot-reloaded by the daemon: a second writer would race the CLI's5+// read-modify-write and thrash the reload throttle. Writes are throttled and6+// dirty-only; last-seen is operational telemetry where minute granularity is7+// plenty.8+9+interface ActivityFile {10+ v: 1;11+ seen: Record<string, string>;12+}13+14+const FLUSH_INTERVAL_MS = 60_000;15+16+/** Daemon-owned last-seen tracking; file-backed when a dataDir is given. */17+export class TicketActivity {18+ #path: string | null;19+ #seen = new Map<string, string>();20+ #dirty = false;21+ #lastFlush = 0;22+ #inFlight: Promise<void> = Promise.resolve();23+24+ private constructor(path: string | null) {25+ this.#path = path;26+ }27+28+ static async load(dataDir?: string): Promise<TicketActivity> {29+ const activity = new TicketActivity(dataDir ? `${dataDir}/ticket-activity.json` : null);30+ if (activity.#path) {31+ try {32+ const parsed = JSON.parse(await Deno.readTextFile(activity.#path)) as ActivityFile;33+ if (parsed.v === 1) {34+ for (const [id, at] of Object.entries(parsed.seen)) activity.#seen.set(id, at);35+ }36+ } catch {37+ // Missing or unreadable sidecar: telemetry starts fresh.38+ }39+ }40+ return activity;41+ }42+43+ /** Record activity on a ticket now; flushes at most once per interval. */44+ note(ticketId: string): void {45+ this.#seen.set(ticketId, new Date().toISOString());46+ this.#dirty = true;47+ if (Date.now() - this.#lastFlush >= FLUSH_INTERVAL_MS) {48+ this.flush().catch(() => {});49+ }50+ }51+52+ /** The last recorded activity for a ticket, if any. */53+ lastSeen(ticketId: string): string | undefined {54+ return this.#seen.get(ticketId);55+ }56+57+ /** Write the sidecar when dirty; called on interval hits and shutdown.58+ * Writes are serialized, and a flush resolves only after every write59+ * scheduled before it has landed. */60+ flush(): Promise<void> {61+ this.#inFlight = this.#inFlight.then(() => this.#write());62+ return this.#inFlight;63+ }64+65+ async #write(): Promise<void> {66+ if (!this.#dirty || !this.#path) return;67+ this.#dirty = false;68+ this.#lastFlush = Date.now();69+ const file: ActivityFile = { v: 1, seen: Object.fromEntries(this.#seen) };70+ const tmp = `${this.#path}.${Deno.pid}.tmp`;71+ await Deno.writeTextFile(tmp, JSON.stringify(file, null, 2) + "\n");72+ await Deno.rename(tmp, this.#path);73+ }74+}75+76+/** Read-only sidecar view for the CLI (a separate process from the daemon). */77+export async function readTicketActivity(dataDir: string): Promise<Map<string, string>> {78+ try {79+ const parsed = JSON.parse(80+ await Deno.readTextFile(`${dataDir}/ticket-activity.json`),81+ ) as ActivityFile;82+ if (parsed.v !== 1) return new Map();83+ return new Map(Object.entries(parsed.seen));84+ } catch {85+ return new Map();86+ }87+}@@ -0,0 +1,93 @@
1+// Connection observability: gate stats carry last-seen, the activity sink is2+// notified on authenticated traffic, and the sidecar round-trips for the CLI.3+4+import { assert, assertEquals } from "@std/assert";5+import { AppTicketStore } from "../src/appticket.ts";6+import { startGate } from "../src/gate.ts";7+import { readTicketActivity, TicketActivity } from "../src/ticket-activity.ts";8+9+function startStubUpstream() {10+ const server = Deno.serve(11+ { hostname: "127.0.0.1", port: 0, onListen: () => {} },12+ () => new Response("{}", { headers: { "content-type": "application/json" } }),13+ );14+ return {15+ url: `http://127.0.0.1:${(server.addr as Deno.NetAddr).port}`,16+ close: () => server.shutdown(),17+ };18+}19+20+Deno.test("authenticated traffic stamps gate stats and the activity sink", async () => {21+ const upstream = startStubUpstream();22+ const store = await AppTicketStore.load();23+ const noted: string[] = [];24+ const gate = startGate({25+ port: 0,26+ hostname: "127.0.0.1",27+ target: () => upstream.url,28+ mode: "ticket",29+ store,30+ appId: "stub-app",31+ adminSecret: "node-admin-secret",32+ activity: { note: (ticketId) => noted.push(ticketId) },33+ });34+ try {35+ const issued = await store.issue("phone");36+ const before = gate.stats();37+ assertEquals(before.length, 0);38+39+ const res = await fetch(`http://127.0.0.1:${gate.port}/t/${issued.secret}/echo`);40+ assertEquals(res.status, 200);41+ await res.body?.cancel();42+43+ const after = gate.stats();44+ const entry = after.find((e) => e.ticketId === issued.record.id);45+ assert(entry, "stats must include the seen ticket");46+ assertEquals(entry.connections, 0);47+ assert(typeof entry.lastSeenAt === "string", "lastSeenAt must be stamped");48+ assert(noted.includes(issued.record.id), "the activity sink must be notified");49+50+ // An invalid secret stamps nothing.51+ const bad = await fetch(`http://127.0.0.1:${gate.port}/t/${"A".repeat(43)}/echo`);52+ assertEquals(bad.status, 401);53+ await bad.body?.cancel();54+ assertEquals(gate.stats().length, 1);55+ } finally {56+ await gate.close();57+ await upstream.close();58+ }59+});60+61+Deno.test("the activity sidecar persists and reloads", async () => {62+ const dir = await Deno.makeTempDir();63+ try {64+ const activity = await TicketActivity.load(dir);65+ assertEquals(activity.lastSeen("t1"), undefined);66+ activity.note("t1");67+ const stamped = activity.lastSeen("t1");68+ assert(typeof stamped === "string", "note must stamp in memory immediately");69+ await activity.flush();70+71+ const viaCli = await readTicketActivity(dir);72+ assertEquals(viaCli.get("t1"), stamped);73+74+ const reloaded = await TicketActivity.load(dir);75+ assertEquals(reloaded.lastSeen("t1"), stamped, "restarts keep last-seen");76+ } finally {77+ await Deno.remove(dir, { recursive: true });78+ }79+});80+81+Deno.test("a missing sidecar reads as empty", async () => {82+ const dir = await Deno.makeTempDir();83+ try {84+ const viaCli = await readTicketActivity(dir);85+ assertEquals(viaCli.size, 0);86+ const memoryOnly = await TicketActivity.load();87+ memoryOnly.note("t1");88+ await memoryOnly.flush();89+ assert(memoryOnly.lastSeen("t1") !== undefined, "memory-only activity still tracks");90+ } finally {91+ await Deno.remove(dir, { recursive: true });92+ }93+});@@ -171,7 +171,8 @@ Deno.test({
171171 ws.addEventListener("open", () => resolve());172172 ws.addEventListener("close", () => reject(new Error("closed before open")));173173 });174- assertEquals(gate.stats(), [{ ticketId: record.id, connections: 1 }]);174+ const live = gate.stats().find((entry) => entry.ticketId === record.id);175+ assertEquals(live?.connections, 1);175176 176177 const closed = new Promise<CloseEvent>((resolve) =>177178 ws.addEventListener("close", (ev) => resolve(ev))@@ -184,7 +185,9 @@ Deno.test({
184185 ),185186 ]);186187 assertEquals(ev.code, CLOSE_TICKET_REVOKED);187- assertEquals(gate.stats(), []);188+ // The socket is gone; the last-seen stamp for the ticket remains.189+ const after = gate.stats().find((entry) => entry.ticketId === record.id);190+ assertEquals(after?.connections, 0);188191 }),189192 });190193