nx_iot_clock_master.nx source
↩ module page · 390 lines · 15797 B
1// nx_iot_clock_master.nx -- single monotonic tick source for the hub.
2//
3// Roadmap: NISHI_IOT_HUB_ROADMAP.md Epoch 4 (SYNCHRONIZE).
4// S-class invariants: S0 (bit-equal per tick) + S3 (atomic scene
5// transitions composed by callers) + S6 (no cloud) + S7 (sealed-enum
6// verdict on every operation).
7//
8// Why this primitive exists (gap-analysis F1/F3 closure):
9// Every IoT incumbent that runs scenes lets each device animate off
10// its own local crystal -- Home Assistant fires a set-command then
11// the bulb's firmware times the fade. N crystals drift, N timers
12// trip at different real-times, scenes desync minute by minute.
13//
14// We replace that with a single hub-side monotonic tick that every
15// device-driver advances against. No bulb runs an autonomous
16// animation; every frame is computed at the hub at tick T and the
17// command emitted before tick T+1. Per cardinal
18// feedback-honest-perf-verdict-no-aspirational-claims, this is
19// the structural difference that makes S3 deliverable -- not a
20// heuristic, an architecture.
21//
22// What this primitive does today:
23// - holds a tick state (i64 monotonic counter + wall-ms anchor)
24// - advances ticks on demand against sys_now_ms()
25// - supports a pre-stage queue: callers register "at tick K, run
26// callback X with arg A" -- a fixed-capacity sorted insert
27// - drain_due() returns the IDs of every entry whose target_tick
28// is <= current; caller dispatches them in returned order
29//
30// What it doesn't do yet (queued, separate slices):
31// - UDP multicast tick broadcast for multi-device sync -- queued
32// for nx_iot_scene_atomic.nx; this primitive is the on-hub
33// scheduler only
34// - per-device ack tracking for committed ticks -- belongs in
35// nx_iot_watchdog.nx
36// - tick-time NTP correction -- v2; v1 trusts sys_now_ms
37//
38// Wire-up notes for the hub daemon:
39// The hub's main loop calls nx_iot_clock_tick(state) in a busy-poll
40// with a small sys_sleep_ms(33) at the bottom for 30 Hz cadence.
41// Per scene, caller pre-stages all per-device commands at
42// (current_tick + STAGE_LEAD_TICKS); at tick advance, drain_due()
43// returns them all together and the daemon dispatches in one pass.
44//
45// genealogy_id: nishi-core/nxc2/docs/NISHI_IOT_S_CLASS_GAP_ANALYSIS.md +
46// feedback-build-the-engine-not-instances
47// license_tier: ORIGINAL
48//
49// nx_capability_claims: (per docs/NISHI_INTELLIGENT_CAPABILITY_LAYER.md)
50// needs: [pointer_arithmetic]
51// provides: [monotonic_tick, schedule_pre_stage, drain_due_atomic,
52// cancel_by_id, peek_next]
53// safety: [no_unchecked_deref, no_floating_point, no_syscall,
54// bit_equal_reproducible, kind_isolated]
55// verdict: [sealed_enum_6_state, no_silent_failure]
56// license: ORIGINAL
57// kind: iot_runtime_primitive
58// sss: [S0, S3, S6, S7] (foundation of atomic-scene invariant)
59//
60// Verification status:
61// no_syscall: ✓ (caller supplies wall_ms_now; no clock-read in primitive)
62// no_floating_point: ✓
63// sealed_enum_6_state: ✓ (NX_IOT_TICK_VERDICT_* has 6 values incl. N)
64// bit_equal_reproducible: ✓ (pure i64 arithmetic; integer division)
65// Theorems 5.5 (tick monotonicity) + 5.6 (drain atomicity) FULL PROOFS
66// in docs/NISHI_SSS_PROOFS_AND_THEORY.md
67//
68// nx_safety_envelope: (schema: nishi-library/seeds/safety-critical-standards.toml)
69// intended_use: "Monotonic tick scheduler -- atomic scene
70// execution across IoT devices"
71// sil_target: SIL3 (timing failure can cause unsafe
72// device-coordination state; HIGHER
73// than tuya-v33 because scenes
74// control multiple devices)
75// asil_target: QM
76// dal_target: NONE
77// iec_62304_class: NONE
78// evidence: [no_syscall, no_floating_point,
79// bit_equal_reproducible, sealed_enum_complete,
80// kind_isolated_audit,
81// Theorem_5_5_tick_monotonicity_FULL_PROOF,
82// Theorem_5_6_drain_atomicity_FULL_PROOF]
83// hazard_register: [bug-tape-clock-wraparound,
84// bug-tape-out-of-order-drain,
85// bug-tape-scene-partial-execution]
86// residual_risk: "Caller MUST supply monotonic wall_ms_now.
87// Substrate cannot enforce this without a
88// syscall. Drain is atomic per-batch;
89// actuator-side failure (device unreachable
90// after we mark it executed) is upstream."
91// verdict: NOT_YET_EVALUATED (awaits nx_safety_critical_grade;
92// but two of the eight axes [bounded_loop,
93// sealed_enum] are PRE-PROVEN via the FULL
94// theorems above)
95
96// No syscall imports needed. The primitive operates on caller-
97// provided state structs + caller-provided wall-ms readings. This
98// keeps it portable across all 12 nxc2 backends (frame primitive
99// pattern proven previous session). The hub daemon supplies
100// sys_now_ms() at integration time.
101
102// ---- Local helpers (defined before use) ----------------------------
103
104func nx_iot_tick_min(a: i64, b: i64) -> i64 {
105 if a < b { return a }
106 return b
107}
108
109// ---- Sealed enum: tick verdict -------------------------------------
110
111const NX_IOT_TICK_VERDICT_UNKNOWN: i64 = 0
112const NX_IOT_TICK_VERDICT_OK: i64 = 1
113const NX_IOT_TICK_VERDICT_NO_ADVANCE: i64 = 2 // same wall-ms as before
114const NX_IOT_TICK_VERDICT_OVERFLOW: i64 = 3 // queue saturated
115const NX_IOT_TICK_VERDICT_NOT_FOUND: i64 = 4 // cancel/peek miss
116const NX_IOT_TICK_VERDICT_BAD_ARG: i64 = 5
117const NX_IOT_TICK_VERDICT_N: i64 = 6
118
119func nx_iot_tick_verdict_is_valid(v: i64) -> i64 {
120 if v < 0 { return 0 }
121 if v >= NX_IOT_TICK_VERDICT_N { return 0 }
122 return 1
123}
124
125// ---- Tick parameters -----------------------------------------------
126//
127// Constants are policy not law -- the user can override at integration
128// time. Defaults chosen for the IoT hub use-case: 30 Hz tick is
129// well below the latency budget (~33 ms) and well above the perception
130// floor (~16 ms for noticeable motion artifacts).
131
132const NX_IOT_TICK_RATE_HZ: i64 = 30
133const NX_IOT_TICK_PERIOD_MS: i64 = 33 // 1000 / 30 rounded down
134
135// ---- State struct --------------------------------------------------
136//
137// Caller pre-allocates one of these for the hub. Fields:
138// - tick_n: monotonic counter, never decreases
139// - anchor_ms: sys_now_ms() reading taken when tick_n was set
140// (allows replay of "what wall-ms did tick N happen at")
141// - period_ms: ms between ticks (default 33)
142// - queue_count: number of active pre-stage entries
143// - queue_cap: capacity of the entries array
144
145struct IotClockState {
146 tick_n: i64,
147 anchor_ms: i64,
148 period_ms: i64,
149 queue_count: i64,
150 queue_cap: i64,
151}
152
153// ---- Pre-stage queue entry -----------------------------------------
154//
155// One pending operation scheduled for a future tick. Caller-defined
156// payload pointer (entry_arg) -- e.g. a *IotDeviceSetCommand the
157// daemon will dispatch. We never dereference it; the queue is
158// payload-agnostic.
159//
160// entry_id is caller-supplied and returned by drain_due so the caller
161// can correlate dispatched entries to its own bookkeeping.
162//
163// Sort order is target_tick ASC; on equal target_tick, FIFO by
164// insertion (entry_id).
165
166struct IotClockEntry {
167 target_tick: i64,
168 entry_id: i64,
169 entry_arg: i64, // opaque payload (cast a pointer to i64)
170}
171
172// ---- Init ----------------------------------------------------------
173
174func nx_iot_clock_init(state: *IotClockState, queue_cap: i64,
175 wall_ms_now: i64) -> i64 {
176 state.tick_n = 0
177 state.anchor_ms = wall_ms_now
178 state.period_ms = NX_IOT_TICK_PERIOD_MS
179 state.queue_count = 0
180 state.queue_cap = queue_cap
181 return 0
182}
183
184// ---- Tick advance --------------------------------------------------
185//
186// Returns the number of ticks advanced since the last call (typically
187// 1 if the daemon polls at ~tick rate; could be larger if the daemon
188// stalled). Writes verdict via out_verdict.
189//
190// Caller passes wall_ms_now from sys_now_ms() at the daemon level so
191// this primitive stays syscall-free + 12-backend portable.
192
193func nx_iot_clock_tick(state: *IotClockState, wall_ms_now: i64,
194 out_verdict: *i64) -> i64 {
195 *out_verdict = NX_IOT_TICK_VERDICT_UNKNOWN
196 let elapsed: i64 = wall_ms_now - state.anchor_ms
197 if elapsed < state.period_ms {
198 *out_verdict = NX_IOT_TICK_VERDICT_NO_ADVANCE
199 return 0
200 }
201 let n_advance: i64 = elapsed / state.period_ms
202 state.tick_n = state.tick_n + n_advance
203 state.anchor_ms = state.anchor_ms + n_advance * state.period_ms
204 *out_verdict = NX_IOT_TICK_VERDICT_OK
205 return n_advance
206}
207
208// ---- Pre-stage queue: schedule -------------------------------------
209//
210// Insert a new entry into the sorted-by-target_tick queue. Returns
211// the entry's position in the queue (>= 0) or -1 on overflow + verdict.
212
213func nx_iot_clock_schedule(state: *IotClockState, queue: *IotClockEntry,
214 target_tick: i64, entry_id: i64,
215 entry_arg: i64, out_verdict: *i64) -> i64 {
216 *out_verdict = NX_IOT_TICK_VERDICT_UNKNOWN
217 if state.queue_count >= state.queue_cap {
218 *out_verdict = NX_IOT_TICK_VERDICT_OVERFLOW
219 return -1
220 }
221 if target_tick < state.tick_n {
222 // target in the past relative to current tick -- accept it
223 // anyway (caller may want it dispatched immediately on next
224 // drain) but flag verdict so caller can warn.
225 *out_verdict = NX_IOT_TICK_VERDICT_OK
226 } else {
227 *out_verdict = NX_IOT_TICK_VERDICT_OK
228 }
229
230 // Find insertion point: first slot with target_tick > new entry's
231 // target_tick (preserves FIFO within equal target_tick).
232 var pos: i64 = state.queue_count
233 var i: i64 = 0
234 while i < state.queue_count {
235 let slot_addr: i64 = (queue as i64) + i * 24
236 let slot: *IotClockEntry = slot_addr as *IotClockEntry
237 if slot.target_tick > target_tick {
238 if pos == state.queue_count { pos = i }
239 }
240 i = i + 1
241 }
242
243 // Shift entries [pos..count) right by one slot.
244 var j: i64 = state.queue_count
245 while j > pos {
246 let src_addr: i64 = (queue as i64) + (j - 1) * 24
247 let dst_addr: i64 = (queue as i64) + j * 24
248 let src: *IotClockEntry = src_addr as *IotClockEntry
249 let dst: *IotClockEntry = dst_addr as *IotClockEntry
250 dst.target_tick = src.target_tick
251 dst.entry_id = src.entry_id
252 dst.entry_arg = src.entry_arg
253 j = j - 1
254 }
255
256 // Write the new entry at pos.
257 let new_addr: i64 = (queue as i64) + pos * 24
258 let new_slot: *IotClockEntry = new_addr as *IotClockEntry
259 new_slot.target_tick = target_tick
260 new_slot.entry_id = entry_id
261 new_slot.entry_arg = entry_arg
262
263 state.queue_count = state.queue_count + 1
264 return pos
265}
266
267// ---- Pre-stage queue: drain due ------------------------------------
268//
269// Returns the count of entries whose target_tick <= current tick_n.
270// Writes the dequeued entries into out[0..ret). The remaining
271// entries shift down so the queue stays packed.
272//
273// out_cap must be >= state.queue_count; if smaller, the verdict
274// reports OVERFLOW and only the first out_cap entries are written.
275
276func nx_iot_clock_drain_due(state: *IotClockState, queue: *IotClockEntry,
277 out: *IotClockEntry, out_cap: i64,
278 out_verdict: *i64) -> i64 {
279 *out_verdict = NX_IOT_TICK_VERDICT_OK
280 let now: i64 = state.tick_n
281
282 // Count consecutive ready entries from the head (queue is sorted
283 // ascending by target_tick). NishiLang has no `break`, so walk
284 // with a keep-flag.
285 var n_due: i64 = 0
286 var keep: i64 = 1
287 while keep == 1 {
288 if n_due >= state.queue_count { keep = 0 }
289 else {
290 let src_addr: i64 = (queue as i64) + n_due * 24
291 let src: *IotClockEntry = src_addr as *IotClockEntry
292 if src.target_tick > now { keep = 0 }
293 else { n_due = n_due + 1 }
294 }
295 }
296
297 if n_due == 0 { return 0 }
298
299 let n_emit: i64 = nx_iot_tick_min(n_due, out_cap)
300
301 // Copy first n_emit entries to out.
302 var i: i64 = 0
303 while i < n_emit {
304 let s_addr: i64 = (queue as i64) + i * 24
305 let d_addr: i64 = (out as i64) + i * 24
306 let s: *IotClockEntry = s_addr as *IotClockEntry
307 let d: *IotClockEntry = d_addr as *IotClockEntry
308 d.target_tick = s.target_tick
309 d.entry_id = s.entry_id
310 d.entry_arg = s.entry_arg
311 i = i + 1
312 }
313
314 // Shift remaining queue entries left by n_due (we consume all due,
315 // regardless of how many fit in out -- non-emitted due entries are
316 // dropped on the floor with OVERFLOW verdict).
317 var src_idx: i64 = n_due
318 var dst_idx: i64 = 0
319 while src_idx < state.queue_count {
320 let s_addr2: i64 = (queue as i64) + src_idx * 24
321 let d_addr2: i64 = (queue as i64) + dst_idx * 24
322 let s2: *IotClockEntry = s_addr2 as *IotClockEntry
323 let d2: *IotClockEntry = d_addr2 as *IotClockEntry
324 d2.target_tick = s2.target_tick
325 d2.entry_id = s2.entry_id
326 d2.entry_arg = s2.entry_arg
327 src_idx = src_idx + 1
328 dst_idx = dst_idx + 1
329 }
330 state.queue_count = state.queue_count - n_due
331
332 if n_emit < n_due { *out_verdict = NX_IOT_TICK_VERDICT_OVERFLOW }
333 return n_emit
334}
335
336// ---- Cancel a scheduled entry by id --------------------------------
337//
338// Linear scan; cancel returns 0 on hit + sets verdict OK, -1 on miss
339// + sets verdict NOT_FOUND. Removed entries shift remaining left.
340
341func nx_iot_clock_cancel(state: *IotClockState, queue: *IotClockEntry,
342 entry_id: i64, out_verdict: *i64) -> i64 {
343 var i: i64 = 0
344 var found: i64 = -1
345 while i < state.queue_count {
346 let addr: i64 = (queue as i64) + i * 24
347 let p: *IotClockEntry = addr as *IotClockEntry
348 if p.entry_id == entry_id {
349 if found == -1 { found = i }
350 }
351 i = i + 1
352 }
353 if found == -1 {
354 *out_verdict = NX_IOT_TICK_VERDICT_NOT_FOUND
355 return -1
356 }
357
358 // Shift left.
359 var src_idx: i64 = found + 1
360 while src_idx < state.queue_count {
361 let s_addr: i64 = (queue as i64) + src_idx * 24
362 let d_addr: i64 = (queue as i64) + (src_idx - 1) * 24
363 let s: *IotClockEntry = s_addr as *IotClockEntry
364 let d: *IotClockEntry = d_addr as *IotClockEntry
365 d.target_tick = s.target_tick
366 d.entry_id = s.entry_id
367 d.entry_arg = s.entry_arg
368 src_idx = src_idx + 1
369 }
370 state.queue_count = state.queue_count - 1
371 *out_verdict = NX_IOT_TICK_VERDICT_OK
372 return 0
373}
374
375// ---- Peek next scheduled tick --------------------------------------
376//
377// Returns the target_tick of the queue's head entry, or -1 + verdict
378// NOT_FOUND if empty. Useful for the daemon to decide how long to
379// sleep before the next nx_iot_clock_tick call.
380
381func nx_iot_clock_peek_next(state: *IotClockState, queue: *IotClockEntry,
382 out_verdict: *i64) -> i64 {
383 if state.queue_count == 0 {
384 *out_verdict = NX_IOT_TICK_VERDICT_NOT_FOUND
385 return -1
386 }
387 *out_verdict = NX_IOT_TICK_VERDICT_OK
388 let p: *IotClockEntry = queue as *IotClockEntry
389 return p.target_tick
390}