OverviewFilesCommitsIssuesDocsReleasesPeers

cthuloid commits / 3eb9244

webui: speak firmware proto 3 (server side of shared control)
Unit: per-key latest-only sending (whole-unit 'look' + one key per slot), grant() matched to its reply (firmware answers in order), release() flushes the slot's pending frame first (else the final value lands after the release and is refused), held/free/values/hello parsed and emitted, 'reset' on reconnect (the unit freed our grants). Rig: for proto-3 units the lamp is the truth -- connect adopts, pushes nothing, no state polling (broadcasts instead). Browser sessions share one socket, which the unit sees as one holder, so the server referees its own sessions before asking the unit. A free's value becomes the slot's stored look. Look recall = grant all, fade, release; units someone holds are skipped and reported. Proto-2 units behave exactly as before. Routes: POST /api/units/:id/grab|release, /slot now carries the session (409 if not held), look recall returns skipped units. Tests: proto-3 fake unit refereeing like grants.h; 56/56 (209 asserts). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
3eb9244f3cd92c26d8d41273750937d78fad4954
by plankatron <fleeky@gmail.com> · 2026-10-02T15:13:07.000Z

Changed files (9) +843 -48

webui/lib/protocol.js +84 -0
……@@ -126,3 +126,87 @@ export function parseGreeting (line) {
126126 const m = /^(\S+) ready$/.exec(line.trim())
127127 return m ? m[1] : null
128128}
129
130// ---- proto 3: shared control (docs/DESIGN-collab.md §5) ---------------------
131//
132// The unit referees per-slot grants. Lines it sends:
133// hello proto=3 unit=<name> epoch=<e> pk=<hex> (line 2 of the greeting)
134// granted <s> <epoch> <n> | busy <s> <holder> (replies to `grant`)
135// held <s> <holder> <epoch> <n> (broadcast)
136// free <s> <epoch> <n> <b> <g> <r> <why> <holder> <sig> (broadcast, signed)
137// values <v1> ... <v12> (broadcast, <= 4/s)
138
139const UINT = /^\d+$/
140
141function uints (parts) {
142 return parts.every((x) => UINT.test(x)) ? parts.map(Number) : null
143}
144
145export function parseHello (line) {
146 const parts = line.trim().split(/\s+/)
147 if (parts[0] !== 'hello') return null
148 const kv = {}
149 for (const p of parts.slice(1)) {
150 const i = p.indexOf('=')
151 if (i > 0) kv[p.slice(0, i)] = p.slice(i + 1)
152 }
153 if (!UINT.test(kv.proto || '') || !UINT.test(kv.epoch || '')) return null
154 return { proto: Number(kv.proto), unit: kv.unit || '', epoch: Number(kv.epoch), pk: kv.pk || '' }
155}
156
157export function parseGranted (line) {
158 const p = line.trim().split(/\s+/)
159 if (p[0] !== 'granted' || p.length !== 4) return null
160 const n = uints(p.slice(1))
161 return n ? { slot: n[0], epoch: n[1], n: n[2] } : null
162}
163
164export function parseBusy (line) {
165 const p = line.trim().split(/\s+/)
166 if (p[0] !== 'busy' || p.length !== 3 || !UINT.test(p[1])) return null
167 return { slot: Number(p[1]), holder: p[2] }
168}
169
170export function parseHeld (line) {
171 const p = line.trim().split(/\s+/)
172 if (p[0] !== 'held' || p.length !== 5) return null
173 const n = uints([p[1], p[3], p[4]])
174 return n ? { slot: n[0], holder: p[2], epoch: n[1], n: n[2] } : null
175}
176
177export function parseFree (line) {
178 const p = line.trim().split(/\s+/)
179 if (p[0] !== 'free' || p.length !== 10) return null
180 const n = uints(p.slice(1, 7))
181 if (!n) return null
182 return { slot: n[0], epoch: n[1], n: n[2], b: n[3], g: n[4], r: n[5], why: p[7], holder: p[8], sig: p[9] }
183}
184
185export function parseValues (line) {
186 const p = line.trim().split(/\s+/)
187 if (p[0] !== 'values' || p.length !== 1 + NUM_CHANNELS) return null
188 return uints(p.slice(1))
189}
190
191export function grantCommand (slot, holder) {
192 return 'grant ' + slot + ' ' + holder + '\n'
193}
194
195export function slotCommand (slot, epoch, n, bgr, fadeMs) {
196 return 'slot ' + slot + ' ' + epoch + ' ' + n + ' ' + bgr.join(' ') + ' ' + Math.trunc(fadeMs) + '\n'
197}
198
199export function releaseCommand (slot) {
200 return 'release ' + slot + '\n'
201}
202
203// The exact bytes a unit signs for a `free` (grants.h freeMessage).
204export function freeMessage (unit, f) {
205 return 'cthuloid/free/1 ' + unit + ' ' + f.slot + ' ' + f.epoch + ' ' + f.n + ' ' + f.b + ' ' + f.g + ' ' + f.r
206}
207
208// Holder ids go to the firmware as one token (<= 32 chars, no spaces).
209export function holderId (name) {
210 const s = String(name || '').trim().replace(/\s+/g, '-').replace(/[^A-Za-z0-9._-]/g, '').slice(0, 32)
211 return s || 'web'
212}
webui/lib/rig.js +155 -12
……@@ -14,6 +14,16 @@
1414// * A unit with no look of ours yet (first connect) ADOPTS what the unit is
1515// actually showing, read back with `state`, instead of pushing black over
1616// whatever Blender or a person left on it.
17//
18// Proto-3 units (shared control, docs/DESIGN-collab.md) change the model:
19// * The LAMP is the truth. Connecting adopts what it shows and pushes
20// nothing; `values` / `free` broadcasts keep us current, so no polling.
21// * Slots are changed only under a grant: grab -> frames -> release. Every
22// browser session of this server shares ONE socket, and the unit sees one
23// socket as one holder, so the server referees its own sessions first and
24// only asks the unit for a slot none of them holds.
25// * A `free` (anyone's release, timeout or disconnect) carries the value the
26// slot ended on; that becomes the slot's stored look.
1727
1828import EventEmitter from 'bare-events'
1929import { Unit } from './unit.js'
……@@ -47,6 +57,12 @@ export function slotsFromValues (values, redMax) {
4757 return slots
4858}
4959
60// One slot from the unit's B, G, R values (as in a `free` broadcast).
61export function slotFromBGR (b, g, r, redMax) {
62 const vals = [b, g, r, 0, 0, 0, 0, 0, 0, 0, 0, 0]
63 return slotsFromValues(vals, redMax)[0]
64}
65
5066export class Rig extends EventEmitter {
5167 constructor ({ store, discovery, readbackMs = 1000, unitOpts = {} }) {
5268 super()
……@@ -103,6 +119,8 @@ export class Rig extends EventEmitter {
103119 slots: Array.isArray(cfg.slots) ? cfg.slots.map(clampSlot) : null,
104120 gamma: cfg.gamma || { b: 1, g: 1, r: 1 },
105121 freq: cfg.freq || 1000,
122 grants: [null, null, null, null], // ours: { session, name, epoch, n, pending }
123 holders: [null, null, null, null], // anyone's, from broadcasts: { holder, epoch, n }
106124 unit: null
107125 }
108126 // Slots saved as all-null are "no look yet", same as never set.
……@@ -110,6 +128,19 @@ export class Rig extends EventEmitter {
110128 entry.unit = new Unit({ id, host: entry.host, port: entry.port, ...this.unitOpts })
111129 entry.unit.on('status', () => this._changed())
112130 entry.unit.on('state', (st) => this._onState(entry, st))
131 entry.unit.on('hello', () => this._changed())
132 entry.unit.on('values', () => this._changed())
133 entry.unit.on('held', (h) => {
134 entry.holders[h.slot] = { holder: h.holder, epoch: h.epoch, n: h.n }
135 this._changed()
136 })
137 entry.unit.on('free', (f) => this._onFree(entry, f))
138 entry.unit.on('reset', () => {
139 // The unit freed every grant of the old socket.
140 entry.grants = [null, null, null, null]
141 entry.holders = [null, null, null, null]
142 this._changed()
143 })
113144 this.units.set(id, entry)
114145 return entry
115146 }
……@@ -136,7 +167,29 @@ export class Rig extends EventEmitter {
136167 this._changed()
137168 }
138169
170 _isV3 (e) {
171 return e.proto >= 3 || e.unit.proto >= 3
172 }
173
174 _onFree (e, f) {
175 const h = e.holders[f.slot]
176 if (h && h.n === f.n && h.epoch === f.epoch) e.holders[f.slot] = null
177 const g = e.grants[f.slot]
178 if (g && !g.pending && g.n === f.n && g.epoch === f.epoch) e.grants[f.slot] = null
179 if (!e.slots) e.slots = [null, null, null, null]
180 e.slots[f.slot] = slotFromBGR(f.b, f.g, f.r, e.redMax)
181 this._save()
182 this._changed()
183 }
184
139185 _onState (e, st) {
186 if (this._isV3(e)) {
187 // The lamp is the truth: always adopt what it shows.
188 e.slots = slotsFromValues(st.values, e.redMax)
189 this._save()
190 this._changed()
191 return
192 }
140193 if (e.slots === null) {
141194 // First sight of this unit's look: adopt it rather than overwrite it.
142195 e.slots = slotsFromValues(st.values, e.redMax)
……@@ -182,11 +235,13 @@ export class Rig extends EventEmitter {
182235 e.gamma = { b: g(patch.gamma.b, e.gamma.b), g: g(patch.gamma.g, e.gamma.g), r: g(patch.gamma.r, e.gamma.r) }
183236 }
184237 if (patch.freq !== undefined) e.freq = P.clamp(Math.trunc(Number(patch.freq) || 1000), 1, 40000)
185 let push = patch.redMax !== undefined
238 // Proto 3: a slots patch updates the model only -- driving the lamp needs a
239 // grant (grab/setSlot/release, or recallLook).
240 let push = patch.redMax !== undefined && !this._isV3(e)
186241 if (Array.isArray(patch.slots)) {
187242 e.slots = patch.slots.slice(0, P.SLOTS_PER_UNIT).map(clampSlot)
188243 while (e.slots.length < P.SLOTS_PER_UNIT) e.slots.push(null)
189 push = true
244 push = !this._isV3(e)
190245 }
191246 if (patch.enabled !== undefined && !!patch.enabled !== e.enabled) {
192247 e.enabled = !!patch.enabled
……@@ -199,10 +254,20 @@ export class Rig extends EventEmitter {
199254 this._changed()
200255 }
201256
202 setSlot (id, index, slot) {
257 setSlot (id, index, slot, session = null) {
203258 const e = this.units.get(id)
204259 if (!e) throw new Error('unknown unit ' + id)
205 if (index < 0 || index >= P.SLOTS_PER_UNIT) throw new Error('slot out of range')
260 if (!(index >= 0 && index < P.SLOTS_PER_UNIT)) throw new Error('slot out of range')
261 if (this._isV3(e)) {
262 const g = e.grants[index]
263 if (!g || g.pending || g.session !== session) throw new Error('not holding slot ' + (index + 1))
264 if (!e.slots) e.slots = [null, null, null, null]
265 e.slots[index] = clampSlot(slot)
266 const s = e.slots[index]
267 e.unit.sendSlot(index, g, P.channelsFromRGB(s.r, s.g, s.b, s.level, e.redMax), e.fadeMs)
268 this._changed()
269 return
270 }
206271 if (!e.slots) e.slots = [null, null, null, null]
207272 e.slots[index] = clampSlot(slot)
208273 this._push(e)
……@@ -210,6 +275,45 @@ export class Rig extends EventEmitter {
210275 this._changed()
211276 }
212277
278 // ---- proto 3 grants ---------------------------------------------------------
279
280 // A browser session asks for a slot. Resolves { granted: true } or
281 // { busy: holderName }.
282 async grab (id, index, session, name) {
283 const e = this.units.get(id)
284 if (!e) throw new Error('unknown unit ' + id)
285 if (!(index >= 0 && index < P.SLOTS_PER_UNIT)) throw new Error('slot out of range')
286 if (!this._isV3(e)) throw new Error('unit has no shared control (firmware proto < 3)')
287 if (!this.connected || !e.enabled || e.unit.status !== 'connected') throw new Error('unit is not connected')
288 const mine = e.grants[index]
289 if (mine) return mine.session === session ? { granted: true } : { busy: mine.name }
290 // Mark it ours BEFORE asking, so a second session grabbing the same slot
291 // in the meantime is refused here rather than racing on the shared socket.
292 const g = { session, name: P.holderId(name), epoch: 0, n: 0, pending: true }
293 e.grants[index] = g
294 const r = await e.unit.grant(index, g.name)
295 if (e.grants[index] !== g) return { busy: 'reset' } // a reconnect wiped it
296 if (r.granted) {
297 g.epoch = r.granted[0].epoch
298 g.n = r.granted[0].n
299 g.pending = false
300 this._changed()
301 return { granted: true }
302 }
303 e.grants[index] = null
304 if (r.busy) return { busy: r.busy.holder }
305 throw new Error(r.error || 'grant failed')
306 }
307
308 release (id, index, session) {
309 const e = this.units.get(id)
310 if (!e) throw new Error('unknown unit ' + id)
311 const g = e.grants[index]
312 if (!g || g.pending || g.session !== session) return false
313 e.unit.release(index) // flushes the slot's pending frame first
314 return true
315 }
316
213317 adoptActual (id) {
214318 const e = this.units.get(id)
215319 if (!e || !e.unit.actual) return false
……@@ -240,12 +344,24 @@ export class Rig extends EventEmitter {
240344
241345 _open (e) {
242346 e.unit.open()
243 if (e.slots === null) e.unit.once('status', () => e.unit.requestState())
244 else this._push(e, { force: true })
347 // Proto is known from mDNS (TXT proto=) or, for units added by hand, from
348 // the `hello` line that follows the greeting.
349 e.unit.once('status', () => this._afterHello(e, () => {
350 if (this._isV3(e) || e.slots === null) e.unit.requestState()
351 else this._push(e, { force: true })
352 }))
353 }
354
355 _afterHello (e, fn) {
356 if (this._isV3(e) || e.unit.proto > 0) return fn()
357 let done = false
358 const go = () => { if (!done) { done = true; clearTimeout(timer); fn() } }
359 const timer = setTimeout(go, 400) // proto <= 2 sends no hello
360 e.unit.once('hello', go)
245361 }
246362
247363 _push (e, { force = false, fadeMs = e.fadeMs } = {}) {
248 if (!this.connected || !e.enabled || e.slots === null) return
364 if (!this.connected || !e.enabled || e.slots === null || this._isV3(e)) return
249365 const values = P.unitValues(e.slots, e.redMax)
250366 const payload = fadeMs > 0 ? P.pwm2Command(values, fadeMs) : P.pwmCommand(values)
251367 e.unit.set(payload, { force })
……@@ -276,17 +392,36 @@ export class Rig extends EventEmitter {
276392 this._changed()
277393 }
278394
279 recallLook (name, fadeMs = 1000) {
395 // Proto 3: take all four slots (or skip the unit if anyone holds one), fade
396 // to the look, let go. The frees then record what the lamp ended on.
397 async recallLook (name, fadeMs = 1000) {
280398 const look = this.looks[name]
281399 if (!look) throw new Error('unknown look ' + name)
400 const skipped = []
282401 for (const [id, slots] of Object.entries(look.units)) {
283402 const e = this.units.get(id)
284403 if (!e) continue
285 e.slots = slots.map(clampSlot)
286 this._push(e, { force: true, fadeMs })
404 if (!this._isV3(e)) {
405 e.slots = slots.map(clampSlot)
406 this._push(e, { force: true, fadeMs })
407 continue
408 }
409 if (!this.connected || !e.enabled || e.unit.status !== 'connected') continue
410 const local = e.grants.find(Boolean)
411 if (local) { skipped.push({ unit: id, holder: local.name }); continue }
412 const r = await e.unit.grant('all', 'look')
413 if (!r.granted) { skipped.push({ unit: id, holder: r.busy ? r.busy.holder : (r.error || '?') }); continue }
414 const want = slots.map(clampSlot)
415 for (const g of r.granted) {
416 const s = want[g.slot]
417 if (!s) continue
418 e.unit.sendSlot(g.slot, g, P.channelsFromRGB(s.r, s.g, s.b, s.level, e.redMax), fadeMs)
419 }
420 e.unit.release('all')
287421 }
288422 this._save()
289423 this._changed()
424 return { skipped }
290425 }
291426
292427 deleteLook (name) {
……@@ -306,7 +441,7 @@ export class Rig extends EventEmitter {
306441 // `state` is cheap but not free on a single-threaded unit: only ask while
307442 // someone is looking, or while a unit's look still has to be adopted.
308443 for (const e of this.units.values()) {
309 if (e.unit.status !== 'connected') continue
444 if (e.unit.status !== 'connected' || this._isV3(e)) continue // v3 broadcasts
310445 if (this.viewers > 0 || e.slots === null) e.unit.requestState()
311446 }
312447 }
……@@ -356,7 +491,15 @@ export class Rig extends EventEmitter {
356491 port: e.port,
357492 online: e.online,
358493 manual: e.manual,
359 proto: e.proto,
494 proto: Math.max(e.proto, e.unit.proto),
495 // Who holds each slot. `session` is set only for this server's own
496 // grants, so a page can tell "mine" from "someone else's".
497 holds: e.holders.map((h, s) => {
498 if (!h) return null
499 const g = e.grants[s]
500 const ours = g && !g.pending && g.n === h.n && g.epoch === h.epoch
501 return { holder: h.holder, session: ours ? g.session : null }
502 }),
360503 enabled: e.enabled,
361504 status: e.unit.status,
362505 error: e.unit.error,
webui/lib/server.js +14 -2
……@@ -104,7 +104,18 @@ export function createServer (rig) {
104104 } else if (kind === 'units' && parts.length === 2) {
105105 rig.updateUnit(id, body)
106106 } else if (kind === 'units' && action === 'slot') {
107 rig.setSlot(id, Number(body.index), body)
107 try {
108 rig.setSlot(id, Number(body.index), body, String(body.session || ''))
109 } catch (err) {
110 if (/not holding/.test(err.message)) throw new HttpError(409, err.message)
111 throw err
112 }
113 } else if (kind === 'units' && action === 'grab') {
114 // Proto 3: a browser session asks for a slot before dragging it.
115 const r = await rig.grab(id, Number(body.index), String(body.session || ''), body.name)
116 return sendJSON(res, r.granted ? 200 : 409, r)
117 } else if (kind === 'units' && action === 'release') {
118 rig.release(id, Number(body.index), String(body.session || ''))
108119 } else if (kind === 'units' && action === 'blackout') {
109120 rig.blackout(id)
110121 } else if (kind === 'units' && action === 'settings') {
……@@ -114,7 +125,8 @@ export function createServer (rig) {
114125 } else if (kind === 'looks' && parts.length === 1) {
115126 rig.saveLook(body.name)
116127 } else if (kind === 'looks' && action === 'recall') {
117 rig.recallLook(id, Number(body.fadeMs) || 0)
128 const r = await rig.recallLook(id, Number(body.fadeMs) || 0)
129 return sendJSON(res, 200, { ok: true, skipped: r.skipped })
118130 } else if (kind === 'looks' && action === 'delete') {
119131 rig.deleteLook(id)
120132 } else {
webui/lib/unit.js +131 -34
……@@ -15,10 +15,22 @@
1515//
1616// Nothing here connects until open() (the Connect switch) or sendOnce()
1717// (Blackout while disconnected).
18//
19// Proto 3 (firmware with shared control) adds, per the design (§5):
20// * per-KEY latest-only sending: 'look' (whole-unit pwm/pwm2) and one key per
21// slot ('slot0'..'slot3'), so a drag on one slot never drops another's frame;
22// * grant() matched to its reply -- the firmware answers commands in order;
23// * release() flushes that slot's pending frame first, or the final value
24// would reach the unit after the release and be refused as not held;
25// * held / free / values broadcasts as events. After a reconnect every grant
26// is gone (the unit freed them on disconnect): 'reset' is emitted and slot
27// frames are never re-asserted, only the 'look'.
1828
1929import EventEmitter from 'bare-events'
2030import net from 'bare-net'
21import { parseGreeting, parseState } from './protocol.js'
31import * as P from './protocol.js'
32
33const GRANT_TIMEOUT_MS = 3000
2234
2335const KEEPALIVE_PAYLOAD = 'hostname\n'
2436
……@@ -46,17 +58,20 @@ export class Unit extends EventEmitter {
4658 this.status = 'off' // off | connecting | connected | unreachable
4759 this.error = ''
4860 this.greetingName = null
49 this.actual = null // last `state` readback
61 this.actual = null // last readback: `state` reply, kept current by `values`
62 this.proto = 0 // from `hello`; 0 until seen (proto <= 2 never sends one)
63 this.epoch = 0
64 this.pk = ''
65 this._grants = [] // pending grant() calls, FIFO: { need, got: [], resolve, timer }
5066
5167 this._wanted = false // open() called and not close()d
5268 this._once = false // sendOnce in progress: deliver, then let go
5369 this._sock = null
5470 this._connected = false
5571 this._buf = ''
56 this._desired = null
57 this._dirty = false
72 this._pending = new Map() // key -> payload not yet written
73 this._lastSent = new Map() // key -> payload last written
5874 this._force = false
59 this._lastSent = null
6075 this._lastTx = 0
6176 this._backoff = backoffStartMs
6277 this._retryTimer = null
……@@ -81,19 +96,69 @@ export class Unit extends EventEmitter {
8196 this._teardown()
8297 clearTimeout(this._retryTimer)
8398 this._retryTimer = null
84 this._lastSent = null
99 this._lastSent.clear()
100 this._pending.clear()
85101 this._setStatus('off', '')
86102 }
87103
88 // Latest look for this unit. Never blocks; delivered when connected.
89 set (payload, { force = false } = {}) {
104 // Latest payload for a key ('look' = the whole unit). Never blocks; delivered
105 // when connected, latest-only per key, identical payloads skipped.
106 set (payload, { force = false, key = 'look' } = {}) {
90107 if (!payload) return
91 this._desired = payload
108 if (!force && this._lastSent.get(key) === payload) {
109 this._pending.delete(key)
110 return
111 }
112 this._pending.set(key, payload)
92113 if (force) this._force = true
93 if (force || payload !== this._lastSent) {
94 this._dirty = true
95 this._flush()
114 this._flush()
115 }
116
117 // ---- proto 3 ------------------------------------------------------------
118
119 // Ask the unit for a slot (0..3) or 'all'. Resolves { granted: [{slot, epoch,
120 // n}, ...] } or { busy: { slot, holder } } or { error }.
121 grant (slot, holder) {
122 return new Promise((resolve) => {
123 if (!this._connected || this.proto < 3) return resolve({ error: 'unit not connected with proto 3' })
124 const req = { need: slot === 'all' ? 4 : 1, got: [], resolve, timer: null }
125 req.timer = setTimeout(() => this._settleGrant(req, { error: 'no reply to grant' }), GRANT_TIMEOUT_MS)
126 this._grants.push(req)
127 this.command(P.grantCommand(slot, holder))
128 })
129 }
130
131 // A drag frame for a held slot: latest-only, rate limited with everything else.
132 sendSlot (slot, grant, bgr, fadeMs) {
133 this.set(P.slotCommand(slot, grant.epoch, grant.n, bgr, fadeMs), { key: 'slot' + slot })
134 }
135
136 // End a grant. Any frame still pending for the slot goes out FIRST.
137 release (slot) {
138 if (!this._connected) return false
139 const slots = slot === 'all' ? [0, 1, 2, 3] : [slot]
140 let out = ''
141 for (const s of slots) {
142 const key = 'slot' + s
143 if (this._pending.has(key)) {
144 out += this._pending.get(key)
145 this._pending.delete(key)
146 }
147 this._lastSent.delete(key)
96148 }
149 return this.command(out + P.releaseCommand(slot))
150 }
151
152 _settleGrant (req, result) {
153 const i = this._grants.indexOf(req)
154 if (i === -1) return
155 this._grants.splice(i, 1)
156 clearTimeout(req.timer)
157 req.resolve(result)
158 }
159
160 _failGrants (error) {
161 for (const req of this._grants.slice()) this._settleGrant(req, { error })
97162 }
98163
99164 // One-shot delivery while not open (Blackout when disconnected): one
……@@ -101,8 +166,7 @@ export class Unit extends EventEmitter {
101166 sendOnce (payload) {
102167 if (this._wanted) return this.set(payload, { force: true })
103168 this._once = true
104 this._desired = payload
105 this._dirty = true
169 this._pending.set('look', payload)
106170 this._force = true
107171 if (!this._sock) this._connect()
108172 else this._flush()
……@@ -142,9 +206,13 @@ export class Unit extends EventEmitter {
142206 this._backoff = this.backoffStartMs
143207 this._lastTx = Date.now()
144208 this._setStatus('connected', '')
145 // The unit may have rebooted or been dropped mid-look; re-assert.
146 this._lastSent = null
147 if (this._desired !== null) this._dirty = true
209 // The unit may have rebooted or been dropped mid-look: re-assert the
210 // look. Slot frames are NOT re-asserted -- their grants died with the
211 // old socket, so the unit would refuse them.
212 const look = this._pending.get('look') || this._lastSent.get('look')
213 this._lastSent.clear()
214 this._pending.clear()
215 if (look) this._pending.set('look', look)
148216 this._flush()
149217 this._armKeepalive()
150218 })
……@@ -157,6 +225,8 @@ export class Unit extends EventEmitter {
157225 if (this._sock !== sock) return
158226 const wasConnected = this._connected
159227 this._teardown()
228 this._failGrants('connection closed')
229 if (wasConnected) this.emit('reset')
160230 if (this._once) {
161231 // One attempt only. A once-send that failed to connect is dropped.
162232 this._once = false
……@@ -165,8 +235,7 @@ export class Unit extends EventEmitter {
165235 }
166236 if (!this._wanted) return
167237 this._setStatus('unreachable', this.error || (wasConnected ? 'connection closed' : 'connect failed'))
168 if (this._desired !== null) this._dirty = true
169 this._lastSent = null
238 for (const k of [...this._pending.keys()]) if (k !== 'look') this._pending.delete(k)
170239 this._retryTimer = setTimeout(() => {
171240 this._retryTimer = null
172241 if (this._wanted && !this._sock) this._connect()
……@@ -194,26 +263,50 @@ export class Unit extends EventEmitter {
194263 const line = this._buf.slice(0, i).trim()
195264 this._buf = this._buf.slice(i + 1)
196265 if (!line) continue
197 const name = parseGreeting(line)
198 if (name !== null) {
199 this.greetingName = name
200 this.emit('greeting', name)
201 continue
202 }
203 const st = parseState(line)
204 if (st !== null) {
205 this.actual = st
206 this.emit('state', st)
207 }
266 this._onLine(line)
208267 }
209268 // A unit that never sends a newline must not grow this without bound.
210269 if (this._buf.length > 4096) this._buf = ''
211270 }
212271
272 _onLine (line) {
273 let x
274 if ((x = P.parseGreeting(line)) !== null) {
275 this.greetingName = x
276 this.emit('greeting', x)
277 } else if ((x = P.parseHello(line)) !== null) {
278 this.proto = x.proto
279 this.epoch = x.epoch
280 this.pk = x.pk
281 this.emit('hello', x)
282 } else if ((x = P.parseState(line)) !== null) {
283 this.actual = x
284 this.emit('state', x)
285 } else if ((x = P.parseValues(line)) !== null) {
286 this.actual = { ...(this.actual || { fading: false, freq: 0, gamma: null }), values: x }
287 this.emit('values', x)
288 } else if ((x = P.parseGranted(line)) !== null) {
289 const req = this._grants[0]
290 if (!req) return
291 req.got.push(x)
292 if (req.got.length >= req.need) this._settleGrant(req, { granted: req.got })
293 } else if ((x = P.parseBusy(line)) !== null) {
294 if (this._grants[0]) this._settleGrant(this._grants[0], { busy: x })
295 } else if ((x = P.parseHeld(line)) !== null) {
296 this.emit('held', x)
297 } else if ((x = P.parseFree(line)) !== null) {
298 this.emit('free', x)
299 } else if (line.startsWith('ERR grant')) {
300 if (this._grants[0]) this._settleGrant(this._grants[0], { error: line })
301 } else if (line.startsWith('ERR notheld')) {
302 this.emit('notheld', line)
303 }
304 }
305
213306 // ---- sending -----------------------------------------------------------
214307
215308 _flush () {
216 if (!this._connected || !this._dirty || this._flushTimer) return
309 if (!this._connected || this._pending.size === 0 || this._flushTimer) return
217310 const wait = this.minIntervalMs - (Date.now() - this._lastTx)
218311 if (!this._force && wait > 0) {
219312 this._flushTimer = setTimeout(() => {
……@@ -222,11 +315,15 @@ export class Unit extends EventEmitter {
222315 }, wait)
223316 return
224317 }
225 const payload = this._desired
318 // Everything pending goes in one write; each key's latest only.
319 let payload = ''
320 for (const [key, p] of this._pending) {
321 payload += p
322 this._lastSent.set(key, p)
323 }
324 this._pending.clear()
226325 this._sock.write(payload)
227326 this._lastTx = Date.now()
228 this._lastSent = payload
229 this._dirty = false
230327 this._force = false
231328 if (this._once) {
232329 // Let the bytes leave before closing; end() flushes, then 'close' fires.
webui/test/helpers.js +125 -0
……@@ -66,3 +66,128 @@ export function waitFor (pred, timeout = 3000) {
6666}
6767
6868export const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms))
69
70// A proto-3 unit: referees per-slot grants like esp324xlight's grants.h, so the
71// web UI's grant handling is tested against the same rules the firmware applies.
72// Signatures are a fixed dummy (signature checks are the bee's job, phase 3).
73export class FakeUnitV3 {
74 constructor (name = 'fake3', { epoch = 5 } = {}) {
75 this.name = name
76 this.epoch = epoch
77 this.counter = 0
78 this.lines = [] // every line received, in order: [connId, line]
79 this.values = new Array(12).fill(0)
80 this.holds = [null, null, null, null] // { conn, holder, n }
81 this.conns = new Map() // id -> socket
82 this.nextId = 1
83 this.server = net.createServer((c) => this._onConn(c))
84 }
85
86 listen () {
87 return new Promise((resolve) => {
88 this.server.listen(0, '127.0.0.1', () => {
89 this.port = this.server.address().port
90 resolve(this)
91 })
92 })
93 }
94
95 close () {
96 for (const c of this.conns.values()) c.destroy()
97 return new Promise((resolve) => this.server.close(resolve))
98 }
99
100 received (pred) {
101 return this.lines.filter(([, l]) => pred(l)).map(([, l]) => l)
102 }
103
104 _broadcast (line) {
105 for (const c of this.conns.values()) c.write(line + '\r\n')
106 }
107
108 _free (s, why) {
109 const h = this.holds[s]
110 this.holds[s] = null
111 const v = this.values.slice(s * 3, s * 3 + 3)
112 this._broadcast(['free', s, this.epoch, h.n, ...v, why, h.holder, 'ab'.repeat(64)].join(' '))
113 }
114
115 _setValues (from, vals, skipHeld) {
116 let changed = false
117 vals.forEach((v, i) => {
118 const c = from + i
119 if (skipHeld && this.holds[Math.floor(c / 3)]) return
120 if (this.values[c] !== v) { this.values[c] = v; changed = true }
121 })
122 if (changed) this._broadcast('values ' + this.values.join(' '))
123 }
124
125 _onConn (c) {
126 const id = this.nextId++
127 this.conns.set(id, c)
128 c.on('error', () => {})
129 // bare-net is half-open by default: 'close' only follows an explicit destroy.
130 c.on('end', () => c.destroy())
131 c.on('close', () => {
132 this.conns.delete(id)
133 for (let s = 0; s < 4; s++) if (this.holds[s] && this.holds[s].conn === id) this._free(s, 'disconnect')
134 })
135 c.write(this.name + ' ready\r\n')
136 c.write('hello proto=3 unit=' + this.name + ' epoch=' + this.epoch + ' pk=' + 'cd'.repeat(32) + '\r\n')
137 let buf = ''
138 c.on('data', (d) => {
139 buf += d.toString()
140 let i
141 while ((i = buf.indexOf('\n')) !== -1) {
142 const line = buf.slice(0, i).trim()
143 buf = buf.slice(i + 1)
144 if (line) this._command(id, c, line)
145 }
146 })
147 }
148
149 _command (id, c, line) {
150 this.lines.push([id, line])
151 const p = line.split(/\s+/)
152 const range = (a) => (a === 'all' ? [0, 1, 2, 3] : [Number(a)])
153 if (p[0] === 'state') {
154 c.write('state 0 1000 1.00 1.00 1.00 ' + this.values.join(' ') + '\r\n')
155 this.holds.forEach((h, s) => { if (h) c.write(['held', s, h.holder, this.epoch, h.n].join(' ') + '\r\n') })
156 c.write('stats epoch=' + this.epoch + ' dropped=0 pk=' + 'cd'.repeat(32) + '\r\n')
157 } else if (p[0] === 'grant') {
158 const slots = range(p[1])
159 const other = slots.find((s) => this.holds[s] && this.holds[s].conn !== id)
160 if (other !== undefined) { c.write('busy ' + other + ' ' + this.holds[other].holder + '\r\n'); return }
161 for (const s of slots) {
162 if (!this.holds[s]) {
163 this.holds[s] = { conn: id, holder: p[2], n: ++this.counter }
164 this._broadcast(['held', s, p[2], this.epoch, this.holds[s].n].join(' '))
165 }
166 c.write(['granted', s, this.epoch, this.holds[s].n].join(' ') + '\r\n')
167 }
168 } else if (p[0] === 'slot') {
169 const [s, e, n, b, g, r] = p.slice(1, 7).map(Number)
170 const h = this.holds[s]
171 if (!h || h.conn !== id || e !== this.epoch || n !== h.n) { c.write('ERR notheld ' + s + '\r\n'); return }
172 this._setValues(s * 3, [b, g, r], false)
173 } else if (p[0] === 'release') {
174 for (const s of range(p[1])) if (this.holds[s] && this.holds[s].conn === id) this._free(s, 'release')
175 } else if (p[0] === 'pwm') {
176 this._setValues(0, p.slice(1).map(Number), true)
177 } else if (p[0] === 'pwm2') {
178 this._setValues(0, p.slice(3).map(Number), true)
179 } else if (p[0] === 'stopfade') {
180 for (let s = 0; s < 4; s++) if (this.holds[s]) this._free(s, 'blackout')
181 }
182 }
183
184 // Another controller (e.g. Blender) grabbing a slot on its own socket.
185 async externalGrab (slot, holder) {
186 const sock = net.createConnection({ port: this.port, host: '127.0.0.1' })
187 await new Promise((resolve) => sock.on('connect', resolve))
188 sock.on('error', () => {})
189 sock.write('grant ' + slot + ' ' + holder + '\n')
190 await waitFor(() => this.holds[slot] && this.holds[slot].holder === holder)
191 return sock
192 }
193}
webui/test/protocol.test.js +32 -0
……@@ -54,3 +54,35 @@ test('state and greeting parsing', (t) => {
5454 t.is(p.parseGreeting('studiolight ready\r\n'), 'studiolight')
5555 t.is(p.parseGreeting('hostname: studiolight'), null)
5656})
57
58// ---- proto 3 (firmware esp324xlight.ino, docs/DESIGN-collab.md §5) ----------
59
60test('proto 3 lines parse', (t) => {
61 t.alike(p.parseHello('hello proto=3 unit=studiolight epoch=22 pk=ab01\r\n'),
62 { proto: 3, unit: 'studiolight', epoch: 22, pk: 'ab01' })
63 t.is(p.parseHello('studiolight ready'), null)
64 t.alike(p.parseHeld('held 2 alice 22 7'), { slot: 2, holder: 'alice', epoch: 22, n: 7 })
65 t.alike(p.parseGranted('granted 0 22 1'), { slot: 0, epoch: 22, n: 1 })
66 t.alike(p.parseBusy('busy 3 bob'), { slot: 3, holder: 'bob' })
67 const sig = 'f'.repeat(128)
68 t.alike(p.parseFree('free 1 22 9 100 200 300 idle carol ' + sig),
69 { slot: 1, epoch: 22, n: 9, b: 100, g: 200, r: 300, why: 'idle', holder: 'carol', sig })
70 t.alike(p.parseValues('values 1 2 3 4 5 6 7 8 9 10 11 12'), [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12])
71 t.is(p.parseValues('values 1 2'), null)
72 t.is(p.parseHeld('held x alice 1 1'), null)
73})
74
75test('proto 3 commands', (t) => {
76 t.is(p.grantCommand(2, 'alice'), 'grant 2 alice\n')
77 t.is(p.grantCommand('all', 'web'), 'grant all web\n')
78 t.is(p.slotCommand(1, 22, 7, [10, 20, 30], 60), 'slot 1 22 7 10 20 30 60\n')
79 t.is(p.releaseCommand(1), 'release 1\n')
80 t.is(p.freeMessage('studiolight', { slot: 2, epoch: 9, n: 41, b: 1, g: 2, r: 45000 }),
81 'cthuloid/free/1 studiolight 2 9 41 1 2 45000')
82})
83
84test('holder ids are made safe for the firmware', (t) => {
85 t.is(p.holderId('Alice Smith!'), 'Alice-Smith')
86 t.is(p.holderId(''), 'web')
87 t.is(p.holderId('x'.repeat(50)).length, 32)
88})
webui/test/rig-v3.test.js +129 -0
……@@ -0,0 +1,129 @@
1// The rig against a proto-3 unit: the server referees its own browser sessions
2// (they share ONE socket, and the unit sees one socket as one holder), and the
3// unit referees everyone else.
4import test from 'brittle'
5import EventEmitter from 'bare-events'
6import { Rig } from '../lib/rig.js'
7import { FakeUnitV3, waitFor, sleep } from './helpers.js'
8
9class FakeDiscovery extends EventEmitter {
10 constructor () { super(); this.available = true; this.reason = '' }
11 start () {}
12 stop () {}
13 announce (fake) {
14 this.emit('up', { name: fake.name, host: fake.name + '.local', addr: '127.0.0.1', port: fake.port, proto: 3 })
15 }
16}
17
18async function setup ({ values, stored, viewers = 0 } = {}) {
19 const fake = await new FakeUnitV3('u3').listen()
20 if (values) fake.values = values
21 const discovery = new FakeDiscovery()
22 const data = stored ? { units: { u3: { enabled: true, fadeMs: 0, slots: stored } } } : {}
23 const rig = new Rig({ store: { load: () => data, save: () => {} }, discovery, readbackMs: 50, unitOpts: { backoffStartMs: 100 } })
24 rig.start()
25 rig.setViewers(viewers)
26 discovery.announce(fake)
27 rig.updateUnit('u3', { enabled: true, fadeMs: 0 })
28 rig.setConnected(true)
29 await waitFor(() => rig.snapshot().units[0].slots !== null)
30 return { fake, rig, teardown: async () => { rig.stop(); await fake.close() } }
31}
32
33const unit = (rig) => rig.snapshot().units[0]
34
35test('connect adopts the lamp, pushes nothing, asks state once', async (t) => {
36 // A stored look AND a page watching: proto 2 would push the look and poll.
37 const white = { r: 1, g: 1, b: 1, level: 1 }
38 const { fake, rig, teardown } = await setup({
39 values: [0, 0, 30000, 0, 0, 0, 0, 0, 0, 0, 0, 0], stored: [white, white, white, white], viewers: 1
40 })
41 await waitFor(() => unit(rig).slots[0].r === 1 && unit(rig).slots[0].g === 0)
42 await sleep(300) // several readback periods
43 t.is(fake.received((l) => l.startsWith('pwm')).length, 0, 'no whole-unit push')
44 t.is(fake.received((l) => l === 'state').length, 1, 'state asked once, then broadcasts')
45 t.is(unit(rig).slots[0].r, 1, 'slot 1 adopted as red')
46 await teardown()
47})
48
49test('sessions are refereed locally; the unit referees other controllers', async (t) => {
50 const { fake, rig, teardown } = await setup()
51 const a = await rig.grab('u3', 0, 'sessA', 'alice')
52 t.ok(a.granted, 'alice grabs slot 1')
53 const b = await rig.grab('u3', 0, 'sessB', 'bob')
54 t.alike(b, { busy: 'alice' }, 'bob is refused by the server itself')
55 t.is(fake.received((l) => l.startsWith('grant 0')).length, 1, 'only one grant reached the unit')
56 t.ok((await rig.grab('u3', 1, 'sessB', 'bob')).granted, 'bob can take another slot')
57
58 const ext = await fake.externalGrab(2, 'blender')
59 t.alike(await rig.grab('u3', 2, 'sessA', 'alice'), { busy: 'blender' }, 'the unit refuses a slot Blender holds')
60 t.ok(await waitFor(() => unit(rig).holds[2] && unit(rig).holds[2].holder === 'blender'), 'holder shown')
61 t.is(unit(rig).holds[2].session, null, 'not one of ours')
62 t.is(unit(rig).holds[0].session, 'sessA')
63 ext.destroy()
64 await teardown()
65})
66
67test('only the holding session can set a slot; frames reach the unit', async (t) => {
68 const { fake, rig, teardown } = await setup()
69 await rig.grab('u3', 1, 'sessA', 'alice')
70 t.exception(() => rig.setSlot('u3', 1, { r: 1, g: 0, b: 0, level: 1 }, 'sessB'), /not holding/)
71 rig.setSlot('u3', 1, { r: 1, g: 0, b: 0, level: 1 }, 'sessA')
72 t.ok(await waitFor(() => fake.values[5] === 30000), 'red at the 30000 cap on slot 2')
73 t.ok(fake.received((l) => l.startsWith('slot 1 5 ')).length >= 1)
74 await teardown()
75})
76
77test('release: the final value is kept and becomes the slot\'s stored look', async (t) => {
78 const { rig, teardown } = await setup()
79 await rig.grab('u3', 3, 'sessA', 'alice')
80 rig.setSlot('u3', 3, { r: 0, g: 0, b: 1, level: 0.5 }, 'sessA')
81 rig.release('u3', 3, 'sessA')
82 t.ok(await waitFor(() => unit(rig).holds[3] === null), 'freed')
83 const s = unit(rig).slots[3]
84 t.ok(Math.abs(s.b - 1) < 1e-6 && Math.abs(s.level - 0.5) < 0.001, 'stored look = what the lamp ended on')
85 await teardown()
86})
87
88test('another controller\'s release updates our view of the slot', async (t) => {
89 const { rig, fake, teardown } = await setup()
90 const ext = await fake.externalGrab(0, 'blender')
91 ext.write('slot 0 5 1 0 65535 0 0\n')
92 ext.write('release 0\n')
93 t.ok(await waitFor(() => unit(rig).slots[0].g === 1 && unit(rig).holds[0] === null), 'slot 1 now green, free')
94 ext.destroy()
95 await teardown()
96})
97
98test('a reconnect clears our grants', async (t) => {
99 const { fake, rig, teardown } = await setup()
100 await rig.grab('u3', 0, 'sessA', 'alice')
101 for (const c of fake.conns.values()) c.destroy()
102 t.ok(await waitFor(() => unit(rig).holds[0] === null), 'grant forgotten')
103 t.exception(() => rig.setSlot('u3', 0, { r: 1, g: 1, b: 1, level: 1 }, 'sessA'), /not holding/)
104 await teardown()
105})
106
107test('recalling a look takes all four slots, fades, and lets go', async (t) => {
108 const { fake, rig, teardown } = await setup()
109 rig.updateUnit('u3', { slots: [{ r: 0, g: 1, b: 0, level: 1 }, null, null, null] })
110 rig.saveLook('green')
111 await rig.recallLook('green', 1500)
112 t.ok(await waitFor(() => fake.received((l) => l === 'release all').length === 1), 'released')
113 t.is(fake.received((l) => l.startsWith('grant all')).length, 1)
114 t.ok(fake.received((l) => /^slot 0 5 \d+ 0 65535 0 1500$/.test(l)).length === 1, 'slot 1 to green over 1.5 s')
115 t.is(fake.received((l) => l.startsWith('pwm')).length, 0, 'no whole-unit pwm')
116 await teardown()
117})
118
119test('recall skips a unit someone else is holding', async (t) => {
120 const { fake, rig, teardown } = await setup()
121 rig.updateUnit('u3', { slots: [{ r: 1, g: 1, b: 1, level: 1 }, null, null, null] })
122 rig.saveLook('white')
123 const ext = await fake.externalGrab(2, 'blender')
124 const r = await rig.recallLook('white', 0)
125 t.alike(r.skipped, [{ unit: 'u3', holder: 'blender' }])
126 t.is(fake.received((l) => l.startsWith('slot ')).length, 0)
127 ext.destroy()
128 await teardown()
129})
webui/test/server-v3.test.js +74 -0
……@@ -0,0 +1,74 @@
1// Real routes, proto-3 unit: grab / slot / release by browser session.
2import test from 'brittle'
3import http from 'bare-http1'
4import EventEmitter from 'bare-events'
5import { Rig } from '../lib/rig.js'
6import { createServer } from '../lib/server.js'
7import { FakeUnitV3, waitFor } from './helpers.js'
8
9class FakeDiscovery extends EventEmitter {
10 constructor () { super(); this.available = true; this.reason = '' }
11 start () {}
12 stop () {}
13}
14
15async function setup () {
16 const fake = await new FakeUnitV3('u3').listen()
17 const discovery = new FakeDiscovery()
18 const rig = new Rig({ store: { load: () => ({}), save: () => {} }, discovery, unitOpts: { backoffStartMs: 100 } })
19 rig.start()
20 discovery.emit('up', { name: 'u3', host: 'u3.local', addr: '127.0.0.1', port: fake.port, proto: 3 })
21 rig.updateUnit('u3', { enabled: true, fadeMs: 0 })
22 rig.setConnected(true)
23 await waitFor(() => rig.snapshot().units[0].slots !== null)
24 const server = createServer(rig)
25 await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve))
26 const port = server.address().port
27 return {
28 fake, rig, port,
29 teardown: async () => { rig.stop(); await new Promise((resolve) => server.close(resolve)); await fake.close() }
30 }
31}
32
33function post (port, path, obj) {
34 return new Promise((resolve, reject) => {
35 const req = http.request({ host: '127.0.0.1', port, method: 'POST', path, headers: { 'content-type': 'application/json', 'x-cthuloid': '1' } }, (res) => {
36 let data = ''
37 res.on('data', (d) => { data += d })
38 res.on('end', () => resolve({ status: res.statusCode, body: data ? JSON.parse(data) : null }))
39 })
40 req.on('error', reject)
41 req.end(JSON.stringify(obj))
42 })
43}
44
45test('grab, drag, release over HTTP; a second session is refused', async (t) => {
46 const s = await setup()
47 const a = await post(s.port, '/api/units/u3/grab', { index: 0, session: 'A', name: 'alice' })
48 t.is(a.status, 200)
49 t.alike(a.body, { granted: true })
50 const b = await post(s.port, '/api/units/u3/grab', { index: 0, session: 'B', name: 'bob' })
51 t.is(b.status, 409)
52 t.alike(b.body, { busy: 'alice' })
53 const notMine = await post(s.port, '/api/units/u3/slot', { index: 0, session: 'B', r: 1, g: 0, b: 0, level: 1 })
54 t.is(notMine.status, 409, 'setting a slot you do not hold is a conflict')
55 const set = await post(s.port, '/api/units/u3/slot', { index: 0, session: 'A', r: 0, g: 1, b: 0, level: 1 })
56 t.is(set.status, 200)
57 t.ok(await waitFor(() => s.fake.values[1] === 65535), 'green reached the unit')
58 const rel = await post(s.port, '/api/units/u3/release', { index: 0, session: 'A' })
59 t.is(rel.status, 200)
60 t.ok(await waitFor(() => s.rig.snapshot().units[0].holds[0] === null), 'released')
61 await s.teardown()
62})
63
64test('look recall reports units it had to skip', async (t) => {
65 const s = await setup()
66 s.rig.updateUnit('u3', { slots: [{ r: 1, g: 1, b: 1, level: 1 }, null, null, null] })
67 s.rig.saveLook('white')
68 const ext = await s.fake.externalGrab(1, 'blender')
69 const r = await post(s.port, '/api/looks/white/recall', { fadeMs: 0 })
70 t.is(r.status, 200)
71 t.alike(r.body.skipped, [{ unit: 'u3', holder: 'blender' }])
72 ext.destroy()
73 await s.teardown()
74})
webui/test/unit-v3.test.js +99 -0
……@@ -0,0 +1,99 @@
1// Unit against a proto-3 unit: grants, per-slot coalescing, release ordering.
2import test from 'brittle'
3import { Unit } from '../lib/unit.js'
4import { FakeUnitV3, waitFor, sleep } from './helpers.js'
5
6async function connected (fake, opts = {}) {
7 const u = new Unit({ id: fake.name, host: '127.0.0.1', port: fake.port, backoffStartMs: 100, ...opts })
8 u.open()
9 await waitFor(() => u.status === 'connected' && u.proto === 3)
10 return u
11}
12
13test('hello: proto, epoch and key are learned', async (t) => {
14 const fake = await new FakeUnitV3('u3', { epoch: 9 }).listen()
15 const u = await connected(fake)
16 t.is(u.proto, 3)
17 t.is(u.epoch, 9)
18 t.is(u.pk, 'cd'.repeat(32))
19 u.close()
20 await fake.close()
21})
22
23test('grant: first come first served, across sockets', async (t) => {
24 const fake = await new FakeUnitV3().listen()
25 const a = await connected(fake)
26 const b = await connected(fake)
27 const ga = await a.grant(1, 'alice')
28 t.alike(ga.granted, [{ slot: 1, epoch: 5, n: 1 }])
29 const gb = await b.grant(1, 'bob')
30 t.alike(gb.busy, { slot: 1, holder: 'alice' })
31 const all = await b.grant('all', 'bob')
32 t.alike(all.busy, { slot: 1, holder: 'alice' }, 'all is atomic')
33 a.close()
34 b.close()
35 await fake.close()
36})
37
38test('slot frames coalesce per slot and never cross slots', async (t) => {
39 const fake = await new FakeUnitV3().listen()
40 const u = await connected(fake, { minIntervalMs: 40 })
41 const g0 = (await u.grant(0, 'alice')).granted[0]
42 const g1 = (await u.grant(1, 'alice')).granted[0]
43 for (let i = 1; i <= 40; i++) {
44 u.sendSlot(0, g0, [i, 0, 0], 0)
45 u.sendSlot(1, g1, [0, i, 0], 0)
46 }
47 t.ok(await waitFor(() => fake.values[0] === 40 && fake.values[4] === 40), 'both slots reach the latest value')
48 const sent = fake.received((l) => l.startsWith('slot '))
49 t.ok(sent.length <= 8, 'intermediate frames dropped: sent ' + sent.length)
50 u.close()
51 await fake.close()
52})
53
54test('release flushes the pending frame first, so the final value is kept', async (t) => {
55 const fake = await new FakeUnitV3().listen()
56 const u = await connected(fake, { minIntervalMs: 1000 }) // a long window: frame still pending
57 const frees = []
58 u.on('free', (f) => frees.push(f))
59 const g = (await u.grant(2, 'alice')).granted[0]
60 u.sendSlot(2, g, [1, 1, 1], 0)
61 u.sendSlot(2, g, [7, 8, 9], 0) // pending behind the rate limit
62 u.release(2)
63 t.ok(await waitFor(() => frees.length === 1), 'free arrived')
64 t.alike([frees[0].b, frees[0].g, frees[0].r], [7, 8, 9], 'free carries the last value, not an older one')
65 t.is(fake.received((l) => l.startsWith('ERR')).length, 0)
66 u.close()
67 await fake.close()
68})
69
70test('broadcasts are emitted: held, free, values', async (t) => {
71 const fake = await new FakeUnitV3().listen()
72 const u = await connected(fake)
73 const seen = { held: [], free: [], values: [] }
74 for (const k of Object.keys(seen)) u.on(k, (x) => seen[k].push(x))
75 const ext = await fake.externalGrab(3, 'blender')
76 t.ok(await waitFor(() => seen.held.length === 1), 'held from another controller')
77 t.is(seen.held[0].holder, 'blender')
78 ext.write('slot 3 5 1 10 20 30 0\n')
79 t.ok(await waitFor(() => seen.values.length > 0 && u.actual.values[9] === 10), 'values update actual')
80 ext.write('release 3\n')
81 t.ok(await waitFor(() => seen.free.length === 1), 'free from another controller')
82 ext.destroy()
83 u.close()
84 await fake.close()
85})
86
87test('a dropped connection fails pending grants and resets', async (t) => {
88 const fake = await new FakeUnitV3().listen()
89 const u = await connected(fake)
90 let resets = 0
91 u.on('reset', () => resets++)
92 const pending = u.grant(0, 'alice')
93 for (const c of fake.conns.values()) c.destroy()
94 const r = await pending
95 t.ok(r.error || r.granted, 'grant settled: ' + JSON.stringify(r))
96 t.ok(await waitFor(() => resets >= 1), 'reset emitted')
97 u.close()
98 await fake.close()
99})