1module Runtime.TelemetryPublication
2
3import Data.Bytes
4import Model.Parameter
5import Model.Word64
6import Data.JSON
7import Observe.Interchange
8import Observe.Stream
9import Observe.Value
10import Runtime.ArenaCertificate
11import Runtime.NativeTelemetry
12import Runtime.TelemetryCapture
13import Runtime.TelemetryIdentity
14import Runtime.TelemetrySeal
15import Std.List
16import Std.Natural
17import Runtime.WordDigest
18
19-- ALPHATEL v2 capture batches reuse the maintained frame and SHA-256 owner.
20-- The identity field contains the four binary AOS identities. Header sequence
21-- is the batch ordinal; counters are the capture control snapshot. Payload is
22-- the prior published source sequence followed by occupied ordinary/reserved
23-- records, each carrying its original source sequence and eight exact u64s.
24-- Monotonic bytes are zero by contract and mean ABSENT, never measured zero.
25-- Status 0 keeps the producer open; 1 declares no further capture records.
26-- This says nothing about the workload's success or a device's clock.
27def publicationMaximumFrameBytes : Nat =
28 65536
29
30def publicationEmptyCounters =
31 (constructor NativeTelemetryCounters NativeTelemetryCountersValue 0 0 0 0 0 0 0 0)
32
33def publicationBodyPrefix : Bytes =
34 (nativeTelemetryRecordBodyForVersion
35 (byte 2)
36 0
37 0
38 (byte 0)
39 (byte 0)
40 (byte 0)
41 (byte 0)
42 publicationEmptyCounters
43 (dataBytesRepeatByte (byte 0) telemetryIdentitySetBytes)
44 b""
45 b""
46 0
47 0)
48
49def publicationPayloadRecordsAt : Nat =
50 (naturalAdd nativeTelemetryEmptyErrorPayloadAt captureWordBytes)
51
52def publicationBodyMaximumBytes =
53 (lambda unrestricted ordinaryCapacity : Nat .
54 (lambda unrestricted reservedCapacity : Nat .
55 (naturalAdd publicationPayloadRecordsAt (captureQueueBytes ordinaryCapacity reservedCapacity))))
56
57def publicationPaddedBytes =
58 (lambda unrestricted bodyBytes : Nat .
59 (naturalMultiply
60 (naturalDivideUnchecked
61 (naturalAdd
62 bodyBytes
63 (naturalAdd 1 (naturalAdd nativeTelemetryScalarBytes (naturalSaturatingSubtract 64 1))))
64 64)
65 64))
66
67def publicationRecordBytes =
68 (lambda unrestricted bodyBytes : Nat .
69 (naturalAdd
70 (bytes-length nativeTelemetryMagic)
71 (naturalAdd bodyBytes nativeTelemetryDigestBytes)))
72
73-- Reclaim requires both durability acknowledgements. A poisoned sink cannot
74-- be retried by appending behind its possibly partial tail. Recovery owns a
75-- new admitted sink/frontier; this primitive never silently discards a batch.
76def publicationReclaimAdmission =
77 (lambda unrestricted version : Nat .
78 (lambda unrestricted writeFull : Nat .
79 (lambda unrestricted fileSynced : Nat .
80 (lambda unrestricted directorySynced : Nat .
81 (lambda unrestricted poisoned : Nat .
82 (aosChoose
83 Nat
84 (naturalAnd
85 (naturalEqual version 1)
86 (naturalAnd
87 (naturalLessOrEqual writeFull 1)
88 (naturalAnd
89 (naturalLessOrEqual fileSynced 1)
90 (naturalAnd
91 (naturalLessOrEqual directorySynced 1)
92 (naturalLessOrEqual poisoned 1)))))
93 (aosChoose
94 Nat
95 poisoned
96 6
97 (aosChoose
98 Nat
99 writeFull
100 (aosChoose Nat (naturalAnd fileSynced directorySynced) 0 3)
101 2))
102 4))))))
103
104-- Capture counts are cumulative except occupied slots. Account for every
105-- selected source sequence since the prior durable frontier before encoding.
106def publicationConservation =
107 (lambda unrestricted priorNext : Nat .
108 (lambda unrestricted priorDropped : Nat .
109 (lambda unrestricted priorExcluded : Nat .
110 (lambda unrestricted next : Nat .
111 (lambda unrestricted dropped : Nat .
112 (lambda unrestricted excluded : Nat .
113 (lambda unrestricted occupied : Nat .
114 (naturalAnd
115 (naturalLessOrEqual priorNext next)
116 (naturalAnd
117 (naturalLessOrEqual priorDropped dropped)
118 (naturalAnd
119 (naturalLessOrEqual priorExcluded excluded)
120 (naturalAnd
121 (naturalLessOrEqual
122 (naturalSaturatingSubtract dropped priorDropped)
123 (naturalSaturatingSubtract next priorNext))
124 (naturalEqual
125 occupied
126 (naturalSaturatingSubtract
127 (naturalSaturatingSubtract next priorNext)
128 (naturalSaturatingSubtract dropped priorDropped))))))))))))))
129
130def publicationSequenceAdmission =
131 (lambda unrestricted lower : Nat .
132 (lambda unrestricted upper : Nat .
133 (lambda unrestricted sequence : Nat .
134 (naturalAnd (naturalLessOrEqual lower sequence) (naturalLess sequence upper)))))
135
136def publicationPublishedNextIndex : Nat =
137 0
138
139def publicationPublishedNextAt : Nat =
140 (naturalMultiply publicationPublishedNextIndex captureWordBytes)
141
142def publicationPublishedDroppedIndex : Nat =
143 (succ publicationPublishedNextIndex)
144
145def publicationPublishedDroppedAt : Nat =
146 (naturalMultiply publicationPublishedDroppedIndex captureWordBytes)
147
148def publicationPublishedExcludedIndex : Nat =
149 (succ publicationPublishedDroppedIndex)
150
151def publicationPublishedExcludedAt : Nat =
152 (naturalMultiply publicationPublishedExcludedIndex captureWordBytes)
153
154def publicationBatchOrdinalIndex : Nat =
155 (succ publicationPublishedExcludedIndex)
156
157def publicationBatchOrdinalAt : Nat =
158 (naturalMultiply publicationBatchOrdinalIndex captureWordBytes)
159
160def publicationFinishedIndex : Nat =
161 (succ publicationBatchOrdinalIndex)
162
163def publicationFinishedAt : Nat =
164 (naturalMultiply publicationFinishedIndex captureWordBytes)
165
166def publicationTerminalRequestIndex : Nat =
167 (succ publicationFinishedIndex)
168
169def publicationTerminalRequestAt : Nat =
170 (naturalMultiply publicationTerminalRequestIndex captureWordBytes)
171
172def publicationBodyLengthIndex : Nat =
173 (succ publicationTerminalRequestIndex)
174
175def publicationBodyLengthAt : Nat =
176 (naturalMultiply publicationBodyLengthIndex captureWordBytes)
177
178def publicationBlockCountIndex : Nat =
179 (succ publicationBodyLengthIndex)
180
181def publicationBlockCountAt : Nat =
182 (naturalMultiply publicationBlockCountIndex captureWordBytes)
183
184def publicationRecordLengthIndex : Nat =
185 (succ publicationBlockCountIndex)
186
187def publicationRecordLengthAt : Nat =
188 (naturalMultiply publicationRecordLengthIndex captureWordBytes)
189
190def publicationPoisonedIndex : Nat =
191 (succ publicationRecordLengthIndex)
192
193def publicationPoisonedAt : Nat =
194 (naturalMultiply publicationPoisonedIndex captureWordBytes)
195
196def publicationReadyIndex : Nat =
197 (succ publicationPoisonedIndex)
198
199def publicationReadyAt : Nat =
200 (naturalMultiply publicationReadyIndex captureWordBytes)
201
202def publicationSavedR12Index : Nat =
203 (succ publicationReadyIndex)
204
205def publicationSavedR12At : Nat =
206 (naturalMultiply publicationSavedR12Index captureWordBytes)
207
208def publicationSavedR13Index : Nat =
209 (succ publicationSavedR12Index)
210
211def publicationSavedR13At : Nat =
212 (naturalMultiply publicationSavedR13Index captureWordBytes)
213
214def publicationSavedR14Index : Nat =
215 (succ publicationSavedR13Index)
216
217def publicationSavedR14At : Nat =
218 (naturalMultiply publicationSavedR14Index captureWordBytes)
219
220def publicationSavedR15Index : Nat =
221 (succ publicationSavedR14Index)
222
223def publicationSavedR15At : Nat =
224 (naturalMultiply publicationSavedR15Index captureWordBytes)
225
226def publicationStateBytes : Nat =
227 (naturalMultiply (succ publicationSavedR15Index) captureWordBytes)
228
229def publicationBodyAt : Nat =
230 (captureAlign publicationStateBytes)
231
232-- The body bound is derived from the payload encoding. Scalar and typed drains
233-- share placement, frame limits, alignment and the same sealing scratch owner.
234def publicationRecordAtForBody =
235 (lambda unrestricted bodyBytes : Nat .
236 (naturalAdd publicationBodyAt (publicationPaddedBytes bodyBytes)))
237
238def publicationZeroClockAtForBody =
239 (lambda unrestricted bodyBytes : Nat .
240 (captureAlign
241 (naturalAdd (publicationRecordAtForBody bodyBytes) (publicationRecordBytes bodyBytes))))
242
243def publicationZeroClockBytes : Nat =
244 (naturalMultiply 2 captureWordBytes)
245
246def publicationSealWorkAtForBody =
247 (lambda unrestricted bodyBytes : Nat .
248 (naturalAdd (publicationZeroClockAtForBody bodyBytes) publicationZeroClockBytes))
249
250def publicationArenaBytesForBody =
251 (lambda unrestricted bodyBytes : Nat .
252 (naturalAdd (publicationSealWorkAtForBody bodyBytes) digestWorkBytes))
253
254def publicationBase =
255 captureWithIdentityArenaBytes
256
257def publicationWholeArenaBytesForBody =
258 (lambda unrestricted applicationBytes : Nat .
259 (lambda unrestricted ordinaryCapacity : Nat .
260 (lambda unrestricted reservedCapacity : Nat .
261 (lambda unrestricted bodyBytes : Nat .
262 (naturalAdd
263 (publicationBase applicationBytes ordinaryCapacity reservedCapacity)
264 (publicationArenaBytesForBody bodyBytes))))))
265
266def publicationWholeArenaResidentsForBody =
267 (lambda unrestricted applicationBytes : Nat .
268 (lambda unrestricted ordinaryCapacity : Nat .
269 (lambda unrestricted reservedCapacity : Nat .
270 (lambda unrestricted bodyBytes : Nat .
271 (constructor
272 ArenaResidents
273 ArenaResidentsNext
274 (constructor
275 ArenaResident
276 ArenaResidentValue
277 b"publication-control"
278 (naturalAdd (publicationBase applicationBytes ordinaryCapacity reservedCapacity) 0)
279 publicationStateBytes
280 captureWordBytes)
281 (constructor
282 ArenaResidents
283 ArenaResidentsNext
284 (constructor
285 ArenaResident
286 ArenaResidentValue
287 b"publication-body"
288 (naturalAdd
289 (publicationBase applicationBytes ordinaryCapacity reservedCapacity)
290 publicationBodyAt)
291 (publicationPaddedBytes bodyBytes)
292 captureWordBytes)
293 (constructor
294 ArenaResidents
295 ArenaResidentsNext
296 (constructor
297 ArenaResident
298 ArenaResidentValue
299 b"publication-record"
300 (naturalAdd
301 (publicationBase applicationBytes ordinaryCapacity reservedCapacity)
302 (publicationRecordAtForBody bodyBytes))
303 (publicationRecordBytes bodyBytes)
304 captureWordBytes)
305 (constructor
306 ArenaResidents
307 ArenaResidentsNext
308 (constructor
309 ArenaResident
310 ArenaResidentValue
311 b"publication-zero-clock"
312 (naturalAdd
313 (publicationBase applicationBytes ordinaryCapacity reservedCapacity)
314 (publicationZeroClockAtForBody bodyBytes))
315 publicationZeroClockBytes
316 captureWordBytes)
317 (constructor
318 ArenaResidents
319 ArenaResidentsNext
320 (constructor
321 ArenaResident
322 ArenaResidentValue
323 b"publication-seal-work"
324 (naturalAdd
325 (publicationBase applicationBytes ordinaryCapacity reservedCapacity)
326 (publicationSealWorkAtForBody bodyBytes))
327 digestWorkBytes
328 captureWordBytes)
329 (captureWithIdentityArenaResidents
330 applicationBytes
331 ordinaryCapacity
332 reservedCapacity))))))))))
333
334def publicationStorageCertificateForBody =
335 (lambda unrestricted applicationBytes : Nat .
336 (lambda unrestricted ordinaryCapacity : Nat .
337 (lambda unrestricted reservedCapacity : Nat .
338 (lambda unrestricted bodyBytes : Nat .
339 (naturalAnd
340 (naturalAnd
341 (naturalEqual telemetryIdentitySetBytes nativeTelemetryDigestBytes)
342 (naturalLessOrEqual publicationPayloadRecordsAt bodyBytes))
343 (naturalAnd
344 (captureWithIdentityArenaCertificate
345 applicationBytes
346 ordinaryCapacity
347 reservedCapacity)
348 (naturalAnd
349 (naturalLessOrEqual (publicationRecordBytes bodyBytes) publicationMaximumFrameBytes)
350 (naturalAnd
351 (naturalLessOrEqual
352 (publicationWholeArenaBytesForBody
353 applicationBytes
354 ordinaryCapacity
355 reservedCapacity
356 bodyBytes)
357 captureExtentMaximum)
358 (arenaCertificate
359 (publicationWholeArenaBytesForBody
360 applicationBytes
361 ordinaryCapacity
362 reservedCapacity
363 bodyBytes)
364 (publicationWholeArenaResidentsForBody
365 applicationBytes
366 ordinaryCapacity
367 reservedCapacity
368 bodyBytes))))))))))
369
370-- Scalar capture is one specialization of the shared bounded body layout.
371def publicationRecordAt =
372 (lambda unrestricted ordinaryCapacity : Nat .
373 (lambda unrestricted reservedCapacity : Nat .
374 (publicationRecordAtForBody (publicationBodyMaximumBytes ordinaryCapacity reservedCapacity))))
375
376def publicationZeroClockAt =
377 (lambda unrestricted ordinaryCapacity : Nat .
378 (lambda unrestricted reservedCapacity : Nat .
379 (publicationZeroClockAtForBody
380 (publicationBodyMaximumBytes ordinaryCapacity reservedCapacity))))
381
382def publicationSealWorkAt =
383 (lambda unrestricted ordinaryCapacity : Nat .
384 (lambda unrestricted reservedCapacity : Nat .
385 (publicationSealWorkAtForBody (publicationBodyMaximumBytes ordinaryCapacity reservedCapacity))))
386
387def publicationArenaBytes =
388 (lambda unrestricted ordinaryCapacity : Nat .
389 (lambda unrestricted reservedCapacity : Nat .
390 (publicationArenaBytesForBody (publicationBodyMaximumBytes ordinaryCapacity reservedCapacity))))
391
392def publicationWholeArenaBytes =
393 (lambda unrestricted applicationBytes : Nat .
394 (lambda unrestricted ordinaryCapacity : Nat .
395 (lambda unrestricted reservedCapacity : Nat .
396 (publicationWholeArenaBytesForBody
397 applicationBytes
398 ordinaryCapacity
399 reservedCapacity
400 (publicationBodyMaximumBytes ordinaryCapacity reservedCapacity)))))
401
402def publicationWholeArenaResidents =
403 (lambda unrestricted applicationBytes : Nat .
404 (lambda unrestricted ordinaryCapacity : Nat .
405 (lambda unrestricted reservedCapacity : Nat .
406 (publicationWholeArenaResidentsForBody
407 applicationBytes
408 ordinaryCapacity
409 reservedCapacity
410 (publicationBodyMaximumBytes ordinaryCapacity reservedCapacity)))))
411
412def publicationStorageCertificate =
413 (lambda unrestricted applicationBytes : Nat .
414 (lambda unrestricted ordinaryCapacity : Nat .
415 (lambda unrestricted reservedCapacity : Nat .
416 (publicationStorageCertificateForBody
417 applicationBytes
418 ordinaryCapacity
419 reservedCapacity
420 (publicationBodyMaximumBytes ordinaryCapacity reservedCapacity)))))
421
422-- Version two carries the existing eight control words. A changed control
423-- shape requires its own admitted encoding, never silent field omission.
424def publicationArenaCertificate =
425 (lambda unrestricted applicationBytes : Nat .
426 (lambda unrestricted ordinaryCapacity : Nat .
427 (lambda unrestricted reservedCapacity : Nat .
428 (naturalAnd
429 (naturalEqual
430 captureControlBytes
431 (bytes-length (nativeTelemetryEncodeCounters publicationEmptyCounters)))
432 (publicationStorageCertificate applicationBytes ordinaryCapacity reservedCapacity)))))
433
434-- Version three is a bounded drain encoding for typed AOS records. Its
435-- identity, control snapshot, ordinal, frontier, digest and closure have the
436-- same owners as version two. The payload is the prior source frontier then
437-- LF-terminated descriptors/events. It does not reinterpret scalar words or
438-- format JSON in acquisition. Original origin/time/availability survive the
439-- drain; schema/stream admission remains Observe.Interchange/Observe.Stream.
440def publicationTypedRecordKey =
441 (lambda unrestricted identities : Bytes .
442 (bytes-append
443 (dataBytesRenderHex
444 (dataBytesTakeValidated
445 telemetryIdentityBytes
446 (dataBytesDropValidated
447 (telemetryIdentityOffset telemetryAttemptIdentityIndex)
448 identities)))
449 (bytes-append
450 b"/"
451 (bytes-append
452 (dataBytesRenderHex
453 (dataBytesTakeValidated
454 telemetryIdentityBytes
455 (dataBytesDropValidated
456 (telemetryIdentityOffset telemetryProducerIdentityIndex)
457 identities)))
458 (bytes-append
459 b"/"
460 (dataBytesRenderHex
461 (dataBytesTakeValidated
462 telemetryIdentityBytes
463 (dataBytesDropValidated
464 (telemetryIdentityOffset telemetryProducerEpochIndex)
465 identities))))))))
466
467def publicationTypedIdentitiesValid =
468 (lambda unrestricted identities : Bytes .
469 (naturalAnd
470 (naturalEqual (bytes-length identities) telemetryIdentitySetBytes)
471 (nat-eliminate
472 (lambda unrestricted current : Nat . Nat)
473 1
474 (lambda unrestricted index : Nat .
475 (lambda unrestricted induction : Nat .
476 (naturalAnd
477 induction
478 (telemetryBinaryIdentityValid
479 (dataBytesTakeValidated
480 telemetryIdentityBytes
481 (dataBytesDropValidated (telemetryIdentityOffset index) identities))))))
482 (succ telemetryProducerEpochIndex))))
483
484def publicationTypedLineAdmission =
485 (lambda unrestricted binding : Bytes .
486 (lambda unrestricted lower : Nat .
487 (lambda unrestricted upper : Nat .
488 (lambda unrestricted line : Bytes .
489 (naturalAnd
490 (naturalIsZero
491 (aosAnyByte (lambda unrestricted b : Byte . (byte-equal b (byte 10))) line))
492 (eliminate
493 AosLine
494 (lambda unrestricted current : (family AosLine) . Nat)
495 (aosClassifyLine line)
496 (branch AosLineInvalid code . 0)
497 (branch
498 AosLineValid
499 record
500 .
501 (aosChoose
502 Nat
503 (bytes-equal (aosMemberBytes b"type" record) b"descriptor")
504 (naturalIsZero (bytes-equal (aosMemberBytes b"kind" record) b"attempt"))
505 (naturalAnd
506 (bytes-equal (aosMemberBytes b"type" record) b"event")
507 (let unrestricted identity =
508 (aosMemberValue b"event_id" record)
509 in
510 (naturalAnd
511 (bytes-equal binding (aosProducerKeyOf identity))
512 (publicationSequenceAdmission
513 lower
514 upper
515 (aosDigitsValue (aosMemberBytes b"sequence" identity))))))))))))))
516
517def publicationTypedRecordBytes =
518 (lambda unrestricted records : (family StdList Bytes) .
519 (bytes-builder-build
520 (eliminate
521 StdList
522 (lambda unrestricted current : (family StdList Bytes) . BytesBuilder)
523 records
524 (branch StdListEmpty . (bytes-builder-empty))
525 (branch
526 StdListCons
527 head
528 tail
529 induction
530 .
531 (bytes-builder-append
532 (bytes-builder-chunk head)
533 (bytes-builder-append (bytes-builder-chunk b"\n") induction))))))
534
535def publicationTypedEventCount =
536 (lambda unrestricted records : (family StdList Bytes) .
537 (eliminate
538 StdList
539 (lambda unrestricted current : (family StdList Bytes) . Nat)
540 records
541 (branch StdListEmpty . 0)
542 (branch
543 StdListCons
544 head
545 tail
546 induction
547 .
548 (naturalAdd
549 induction
550 (eliminate
551 AosLine
552 (lambda unrestricted current : (family AosLine) . Nat)
553 (aosClassifyLine head)
554 (branch AosLineInvalid code . 0)
555 (branch AosLineValid record . (bytes-equal (aosMemberBytes b"type" record) b"event")))))))
556
557-- This checks individual record binding and the bounded payload population.
558-- Cross-record sequence uniqueness, cumulative conservation, descriptor
559-- immutability and stream schema meaning are additional admission gates; a
560-- successful result here is not a publication/durability acknowledgement.
561def publicationTypedPayloadAdmission =
562 (lambda unrestricted identities : Bytes .
563 (lambda unrestricted lower : Nat .
564 (lambda unrestricted upper : Nat .
565 (lambda unrestricted occupied : Nat .
566 (lambda unrestricted records : (family StdList Bytes) .
567 (naturalAnd
568 (publicationTypedIdentitiesValid identities)
569 (naturalAnd
570 (naturalEqual (publicationTypedEventCount records) occupied)
571 (naturalAnd
572 (naturalLessOrEqual (aosListLength records) 1024)
573 (naturalAnd
574 (naturalLessOrEqual
575 (publicationRecordBytes
576 (naturalAdd
577 publicationPayloadRecordsAt
578 (bytes-length (publicationTypedRecordBytes records))))
579 publicationMaximumFrameBytes)
580 (eliminate
581 StdList
582 (lambda unrestricted current : (family StdList Bytes) . Nat)
583 records
584 (branch StdListEmpty . 1)
585 (branch
586 StdListCons
587 head
588 tail
589 induction
590 .
591 (naturalAnd
592 induction
593 (publicationTypedLineAdmission
594 (publicationTypedRecordKey identities)
595 lower
596 upper
597 head)))))))))))))
598
599-- A low-level body constructor, like nativeTelemetryRecordBodyForVersion.
600-- The caller must admit its control snapshot and typed payload before seal;
601-- the strict collector independently rechecks both against the original bytes.
602def publicationTypedBody =
603 (lambda unrestricted ordinal : (family ModelWord64) .
604 (lambda unrestricted terminal : Byte .
605 (lambda unrestricted counters : (family NativeTelemetryCounters) .
606 (lambda unrestricted identities : Bytes .
607 (lambda unrestricted priorNext : (family ModelWord64) .
608 (lambda unrestricted records : (family StdList Bytes) .
609 (let unrestricted payload =
610 (bytes-append
611 (nativeTelemetryWord64Bytes priorNext)
612 (publicationTypedRecordBytes records))
613 in
614 (nativeTelemetryRecordBodyForVersion
615 (byte 3)
616 ordinal
617 nativeTelemetryZeroWord64
618 (byte 0)
619 (byte 0)
620 (byte 0)
621 terminal
622 counters
623 identities
624 b""
625 payload
626 nativeTelemetryZeroWord64
627 (modelWord64FromNaturalTruncated (bytes-length payload))))))))))The compiler supplied declaration spans and resolved links from this source snapshot. This page does not assert that this file belongs to a checked closure.