Public: anyone can read this; the crew can change it.
"/system over server-sent events > the sampler's clock runs only while someone watches, and ticks while they do" passes every time on a loaded check machine #5478
Check · Review · Land · Live done — Landed and live in production.
scout/forge:scout/5473-system-over-server-sent-events-the-sampl → main 1 commit fork 1 thread · 1 unresolved
2 files
changed +31
−11
src/system/live.ts +14 −5
| @@ -28,7 +28,16 @@ export type Live = { | ||
| 28 | 28 | publish(view: SystemView): void; |
| 29 | 29 | }; |
| 30 | 30 | |
| 31 | export function createLive(options: { paths?: Paths; roots?: () => Roots; self?: number; pushMs?: number } = {}): Live { | |
| 31 | /** Run `tick` every `ms` until the returned function is called; a test injects its own. */ | |
| 32 | export type Schedule = (tick: () => void, ms: number) => () => void; | |
| 33 | ||
| 34 | const realSchedule: Schedule = (tick, ms) => { | |
| 35 | const t = setInterval(tick, ms); | |
| 36 | t.unref?.(); | |
| 37 | return () => clearInterval(t); | |
| 38 | }; | |
| 39 | ||
| 40 | export function createLive(options: { paths?: Paths; roots?: () => Roots; self?: number; pushMs?: number; schedule?: Schedule } = {}): Live { | |
| 32 | 41 | const paths = options.paths ?? HOST; |
| 33 | 42 | const roots = options.roots ?? (() => cellRoots()); |
| 34 | 43 | const self = options.self ?? process.pid; |
| @@ -37,18 +46,18 @@ export function createLive(options: { paths?: Paths; roots?: () => Roots; self?: | ||
| 37 | 46 | let view: SystemView | undefined; |
| 38 | 47 | let history: number[][] = []; |
| 39 | 48 | const listeners = new Set<(view: SystemView) => void>(); |
| 40 | let timer: ReturnType<typeof setInterval> | undefined; | |
| 49 | const schedule = options.schedule ?? realSchedule; | |
| 50 | let timer: (() => void) | undefined; | |
| 41 | 51 | const self_: Live = { |
| 42 | 52 | subscribe(listener) { |
| 43 | 53 | listeners.add(listener); |
| 44 | 54 | if (timer === undefined) { |
| 45 | timer = setInterval(() => self_.publish(self_.read()), options.pushMs ?? PUSH_MS); | |
| 46 | timer.unref?.(); | |
| 55 | timer = schedule(() => self_.publish(self_.read()), options.pushMs ?? PUSH_MS); | |
| 47 | 56 | } |
| 48 | 57 | return () => { |
| 49 | 58 | listeners.delete(listener); |
| 50 | 59 | if (listeners.size === 0 && timer !== undefined) { |
| 51 | clearInterval(timer); | |
| 60 | timer(); | |
| 52 | 61 | timer = undefined; |
| 53 | 62 | } |
| 54 | 63 | }; |
test/system-stream.test.ts +17 −6
| @@ -100,21 +100,32 @@ describe("/system over server-sent events", () => { | ||
| 100 | 100 | }); |
| 101 | 101 | |
| 102 | 102 | test("the sampler's clock runs only while someone watches, and ticks while they do", async () => { |
| 103 | const own = createLive({ pushMs: 10 }); | |
| 103 | // A hand-cranked clock: no sleeping, so a loaded machine cannot starve it. | |
| 104 | let tick: (() => void) | undefined; | |
| 105 | let cancelled = 0; | |
| 106 | const own = createLive({ | |
| 107 | schedule: (fn) => { | |
| 108 | tick = fn; | |
| 109 | return () => { | |
| 110 | tick = undefined; | |
| 111 | cancelled++; | |
| 112 | }; | |
| 113 | }, | |
| 114 | }); | |
| 104 | 115 | expect(own.running()).toBe(false); |
| 105 | 116 | const first = own.subscribe(() => {}); |
| 106 | 117 | expect(own.running()).toBe(true); |
| 107 | 118 | first(); |
| 108 | 119 | expect(own.running()).toBe(false); |
| 120 | expect(cancelled).toBe(1); | |
| 109 | 121 | let ticks = 0; |
| 110 | 122 | const second = own.subscribe(() => ticks++); |
| 111 | 123 | expect(own.running()).toBe(true); |
| 112 | await Bun.sleep(100); | |
| 124 | for (let i = 0; i < 3; i++) tick!(); | |
| 125 | expect(ticks).toBe(3); | |
| 113 | 126 | second(); |
| 114 | expect(ticks).toBeGreaterThan(2); | |
| 115 | const stopped = ticks; | |
| 116 | await Bun.sleep(60); | |
| 117 | expect(ticks).toBe(stopped); | |
| 127 | expect(tick).toBeUndefined(); | |
| 128 | expect(cancelled).toBe(2); | |
| 118 | 129 | expect(own.running()).toBe(false); |
| 119 | 130 | }); |
| 120 | 131 |
new above
offline
Nit: the
Scheduletype is exported only for tests. That is fine, but a one-line note that the default is unref-ed would help the next reader.