lib/machine/src/fabric/snapshot/decode.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const admission = @import("../../admission/root.zig");
2 const fabric = @import("../root.zig");
3 const os = @import("os");
4 const schema = @import("schema.zig");
5 const snapshot_types = @import("types.zig");
6 const std = @import("std");
7
8 const State = @FieldType(fabric.Fabric, "state");
9 const NodeState = @typeInfo(@FieldType(State, "nodes")).array.child;
10 const QueuedPacket = @typeInfo(@FieldType(State, "packets")).array.child;
11 const wire = os.abi.wire;
12
13 pub fn snapshot(
14 input: *const [schema.stream_bytes]u8,
15 expected_root: fabric.Root,
16 ) snapshot_types.Error!fabric.Fabric {
17 try validateEnvelope(input);
18 const node_count = input[schema.StateLayout.node_count_offset];
19 const packet_count = input[schema.StateLayout.packet_count_offset];
20 try validateMaterialShape(input, node_count, packet_count);
21 std.debug.assert(node_count <= fabric.node_limit);
22 std.debug.assert(packet_count <= fabric.packet_limit);
23 try validateReserved(input, node_count, packet_count);
24 const root_value = try decodeRoot(input);
25 if (!std.meta.eql(root_value, expected_root)) return error.FabricRootMismatch;
26 var state = decodeState(input, root_value, node_count, packet_count);
27 for (&state.nodes, 0..) |*node, index| {
28 if (index == node_count) break;
29 node.* = try decodeNode(input, index);
30 }
31 for (&state.packets, 0..) |*packet, index| {
32 if (index == packet_count) break;
33 packet.* = try decodePacket(input, index);
34 }
35 const result: fabric.Fabric = .{ .state = state };
36 _ = try result.cut();
37 std.debug.assert(std.meta.eql(result.root(), expected_root));
38 return result;
39 }
40
41 fn validateEnvelope(
42 input: *const [schema.stream_bytes]u8,
43 ) snapshot_types.Error!void {
44 const layout = schema.Envelope;
45 if (!std.mem.eql(
46 u8,
47 input[layout.magic_offset..layout.version_offset],
48 &schema.magic,
49 )) return error.BadMagic;
50 if (get16(input, layout.version_offset) != schema.version) {
51 return error.UnsupportedVersion;
52 }
53 if (get16(input, layout.stream_bytes_offset) != schema.stream_bytes) {
54 return error.StreamBytesMismatch;
55 }
56 if (get16(input, layout.flags_offset) != schema.flags) {
57 return error.UnsupportedFlags;
58 }
59 if (get16(input, layout.format_offset) != schema.format) {
60 return error.SnapshotFormatMismatch;
61 }
62 if (get16(input, layout.dialect_offset) !=
63 @backingInt(fabric.Dialect.ordered_effect_fabric_v3))
64 {
65 return error.UnknownSnapshotField;
66 }
67 try validateEnvelopeShape(input);
68 }
69
70 fn validateEnvelopeShape(
71 input: *const [schema.stream_bytes]u8,
72 ) snapshot_types.Error!void {
73 const layout = schema.Envelope;
74 if (get16(input, layout.node_capacity_offset) != fabric.node_limit or
75 get16(input, layout.packet_capacity_offset) != fabric.packet_limit or
76 get16(input, layout.node_bytes_offset) != schema.node_bytes or
77 get16(input, layout.packet_bytes_offset) != schema.packet_bytes or
78 get16(input, layout.packet_payload_bytes_offset) !=
79 schema.packet_payload_bytes or
80 get16(input, layout.digest_bytes_offset) != schema.digest_bytes or
81 get16(input, layout.root_bytes_offset) != schema.root_bytes or
82 get16(input, layout.state_header_bytes_offset) !=
83 schema.state_header_bytes)
84 {
85 return error.MetadataShapeMismatch;
86 }
87 }
88
89 fn validateMaterialShape(
90 input: *const [schema.stream_bytes]u8,
91 node_count: u8,
92 packet_count: u8,
93 ) snapshot_types.Error!void {
94 if (node_count == 0 or node_count > fabric.node_limit or
95 packet_count > fabric.packet_limit)
96 {
97 return error.MetadataShapeMismatch;
98 }
99 for (0..packet_count) |index| {
100 const start = schema.packets_offset + index * schema.packet_bytes;
101 const length = get16(input, start + schema.PacketLayout.length_offset);
102 if (length == 0 or length > schema.packet_payload_bytes) {
103 return error.MetadataShapeMismatch;
104 }
105 }
106 }
107
108 fn validateReserved(
109 input: *const [schema.stream_bytes]u8,
110 node_count: u8,
111 packet_count: u8,
112 ) snapshot_types.Error!void {
113 if (!zeroRange(input, schema.Envelope.reserved_offset, schema.Envelope.limit) or
114 !zeroRange(input, schema.RootLayout.reserved_offset, schema.RootLayout.limit) or
115 !zeroRange(
116 input,
117 schema.StateLayout.first_reserved_offset,
118 schema.StateLayout.virtual_time_offset,
119 ) or
120 !zeroRange(
121 input,
122 schema.StateLayout.second_reserved_offset,
123 schema.StateLayout.ledger_digest_offset,
124 ) or
125 !zeroRange(
126 input,
127 schema.StateLayout.third_reserved_offset,
128 schema.StateLayout.limit,
129 ) or
130 !wire.allZero(input[schema.material_limit..]))
131 {
132 return error.ReservedNonzero;
133 }
134 try validateNodeReserved(input, node_count);
135 try validatePacketReserved(input, packet_count);
136 }
137
138 fn validateNodeReserved(
139 input: *const [schema.stream_bytes]u8,
140 node_count: u8,
141 ) snapshot_types.Error!void {
142 const layout = schema.NodeLayout;
143 for (0..fabric.node_limit) |index| {
144 const start = schema.nodes_offset + index * schema.node_bytes;
145 if (index >= node_count) {
146 if (!wire.allZero(input[start..][0..schema.node_bytes])) {
147 return error.ReservedNonzero;
148 }
149 continue;
150 }
151 if (!zeroRange(
152 input,
153 start + layout.first_reserved_offset,
154 start + layout.request_receipt_offset,
155 ) or !zeroRange(
156 input,
157 start + layout.second_reserved_offset,
158 start + layout.machine_digest_offset,
159 )) {
160 return error.ReservedNonzero;
161 }
162 if (input[start + layout.request_present_offset] == 0 and
163 !zeroRange(
164 input,
165 start + layout.request_receipt_offset,
166 start + layout.machine_kind_offset,
167 ))
168 {
169 return error.ReservedNonzero;
170 }
171 }
172 }
173
174 fn validatePacketReserved(
175 input: *const [schema.stream_bytes]u8,
176 packet_count: u8,
177 ) snapshot_types.Error!void {
178 const layout = schema.PacketLayout;
179 for (0..fabric.packet_limit) |index| {
180 const start = schema.packets_offset + index * schema.packet_bytes;
181 if (index >= packet_count) {
182 if (!wire.allZero(input[start..][0..schema.packet_bytes])) {
183 return error.ReservedNonzero;
184 }
185 continue;
186 }
187 const length = get16(input, start + layout.length_offset);
188 if (!zeroRange(
189 input,
190 start + layout.reserved_offset,
191 start + layout.request_receipt_offset,
192 ) or
193 !wire.allZero(input[start + layout.storage_offset + length .. start + layout.limit]))
194 {
195 return error.ReservedNonzero;
196 }
197 }
198 }
199
200 fn decodeRoot(
201 input: *const [schema.stream_bytes]u8,
202 ) snapshot_types.Error!fabric.Root {
203 const layout = schema.RootLayout;
204 return .{
205 .digest = digest(input, layout.digest_offset),
206 .dialect = try enumValue(fabric.Dialect, get16(input, layout.dialect_offset)),
207 .machine_contract = .{ .digest = digest(input, layout.contract_offset) },
208 .entry_frontier = get64(input, layout.entry_frontier_offset),
209 .admission_frontier = get64(input, layout.admission_frontier_offset),
210 .fault_frontier = get64(input, layout.fault_frontier_offset),
211 };
212 }
213
214 fn decodeState(
215 input: *const [schema.stream_bytes]u8,
216 root_value: fabric.Root,
217 node_count: u8,
218 packet_count: u8,
219 ) State {
220 const layout = schema.StateLayout;
221 return .{
222 .contract = root_value.machine_contract,
223 .world = input[layout.world_offset..][0..16].*,
224 .node_count = node_count,
225 .nodes = @splat(emptyNode()),
226 .virtual_time_tick = get64(input, layout.virtual_time_offset),
227 .entropy_frontier = get64(input, layout.entropy_frontier_offset),
228 .entropy_bytes = get64(input, layout.entropy_bytes_offset),
229 .packet_next_id = get64(input, layout.packet_next_id_offset),
230 .packet_count = packet_count,
231 .packets = @splat(emptyPacket()),
232 .partition_bits = input[layout.partition_bits_offset],
233 .capacity_bits = input[layout.capacity_bits_offset],
234 .pending = null,
235 .entry_frontier = get64(input, layout.entry_frontier_offset),
236 .admission_frontier = get64(input, layout.admission_frontier_offset),
237 .fault_frontier = get64(input, layout.fault_frontier_offset),
238 .admission_limit = get32(input, layout.admission_limit_offset),
239 .ledger_digest = digest(input, layout.ledger_digest_offset),
240 .fault_digest = digest(input, layout.fault_digest_offset),
241 .root = root_value,
242 };
243 }
244
245 fn decodeNode(
246 input: *const [schema.stream_bytes]u8,
247 index: usize,
248 ) snapshot_types.Error!NodeState {
249 std.debug.assert(index < fabric.node_limit);
250 const layout = schema.NodeLayout;
251 const start = schema.nodes_offset + index * schema.node_bytes;
252 return .{
253 .id = .{ .bytes = input[start + layout.id_offset ..][0..16].* },
254 .basis = try decodeBasis(input, start),
255 .machine = .{
256 .kind = try enumValue(
257 fabric.MachineBoundaryKind,
258 input[start + layout.machine_kind_offset],
259 ),
260 .digest = digest(input, start + layout.machine_digest_offset),
261 },
262 .available = try boolean(input[start + layout.available_offset]),
263 };
264 }
265
266 fn decodeBasis(
267 input: *const [schema.stream_bytes]u8,
268 start: usize,
269 ) snapshot_types.Error!admission.Basis {
270 std.debug.assert(start >= schema.nodes_offset);
271 std.debug.assert(start + schema.node_bytes <= schema.packets_offset);
272 const layout = schema.NodeLayout;
273 const present = try boolean(input[start + layout.request_present_offset]);
274 const request: ?admission.EffectRequest = if (present) .{
275 .receipt = .{ .digest = digest(input, start + layout.request_receipt_offset) },
276 .correlation = get64(input, start + layout.request_correlation_offset),
277 } else null;
278 return .{
279 .contract = .{ .digest = digest(input, start + layout.contract_offset) },
280 .source_root = digest(input, start + layout.source_root_offset),
281 .frontiers = .{
282 .input = get64(input, start + layout.input_frontier_offset),
283 .terminal_input_offset = get64(input, start + layout.terminal_input_offset),
284 .virtual_time_tick = get64(input, start + layout.virtual_time_offset),
285 .entropy_generation = get64(input, start + layout.entropy_offset),
286 .effect = get64(input, start + layout.effect_offset),
287 },
288 .outstanding_effect = request,
289 };
290 }
291
292 fn decodePacket(
293 input: *const [schema.stream_bytes]u8,
294 index: usize,
295 ) snapshot_types.Error!QueuedPacket {
296 std.debug.assert(index < fabric.packet_limit);
297 const layout = schema.PacketLayout;
298 const start = schema.packets_offset + index * schema.packet_bytes;
299 const length = get16(input, start + layout.length_offset);
300 std.debug.assert(length > 0);
301 std.debug.assert(length <= fabric.packet_bytes_max);
302 var storage: [fabric.packet_bytes_max]u8 = @splat(0);
303 @memcpy(storage[0..length], input[start + layout.storage_offset ..][0..length]);
304 return .{
305 .packet = .{
306 .id = get64(input, start + layout.id_offset),
307 .source = .{ .bytes = input[start + layout.source_offset ..][0..16].* },
308 .destination = .{
309 .bytes = input[start + layout.destination_offset ..][0..16].*,
310 },
311 .channel = get16(input, start + layout.channel_offset),
312 .request = .{
313 .receipt = .{
314 .digest = digest(input, start + layout.request_receipt_offset),
315 },
316 .correlation = get64(input, start + layout.request_correlation_offset),
317 },
318 .length = length,
319 .storage = storage,
320 .digest = digest(input, start + layout.digest_offset),
321 },
322 .ready = try boolean(input[start + layout.ready_offset]),
323 .ready_at_tick = get64(input, start + layout.ready_at_offset),
324 .order = get64(input, start + layout.order_offset),
325 };
326 }
327
328 fn emptyNode() NodeState {
329 return .{
330 .id = .{ .bytes = @splat(0) },
331 .basis = .{
332 .contract = .{ .digest = @splat(0) },
333 .source_root = @splat(0),
334 .frontiers = .{
335 .input = 0,
336 .terminal_input_offset = 0,
337 .virtual_time_tick = 0,
338 .entropy_generation = 0,
339 .effect = 0,
340 },
341 .outstanding_effect = null,
342 },
343 .machine = .{ .kind = .basis, .digest = @splat(0) },
344 .available = false,
345 };
346 }
347
348 fn emptyPacket() QueuedPacket {
349 return .{
350 .packet = .{
351 .id = 0,
352 .source = .{ .bytes = @splat(0) },
353 .destination = .{ .bytes = @splat(0) },
354 .channel = 0,
355 .request = .{
356 .receipt = .{ .digest = @splat(0) },
357 .correlation = 0,
358 },
359 .length = 0,
360 .storage = @splat(0),
361 .digest = @splat(0),
362 },
363 .ready = false,
364 .ready_at_tick = 0,
365 .order = 0,
366 };
367 }
368
369 fn boolean(value: u8) snapshot_types.Error!bool {
370 return switch (value) {
371 0 => false,
372 1 => true,
373 else => error.NonCanonicalBoolean,
374 };
375 }
376
377 fn enumValue(comptime T: type, value: anytype) snapshot_types.Error!T {
378 inline for (std.meta.tags(T)) |candidate| {
379 if (@backingInt(candidate) == value) return candidate;
380 }
381 return error.UnknownSnapshotField;
382 }
383
384 fn digest(input: *const [schema.stream_bytes]u8, offset: usize) os.abi.Digest {
385 std.debug.assert(offset + schema.digest_bytes <= input.len);
386 return input[offset..][0..schema.digest_bytes].*;
387 }
388
389 fn zeroRange(
390 input: *const [schema.stream_bytes]u8,
391 first: usize,
392 last: usize,
393 ) bool {
394 std.debug.assert(first <= last);
395 std.debug.assert(last <= input.len);
396 return wire.allZero(input[first..last]);
397 }
398
399 fn get16(input: *const [schema.stream_bytes]u8, offset: usize) u16 {
400 std.debug.assert(offset + 2 <= input.len);
401 return wire.read16(input[offset..][0..2]);
402 }
403
404 fn get32(input: *const [schema.stream_bytes]u8, offset: usize) u32 {
405 std.debug.assert(offset + 4 <= input.len);
406 return wire.read32(input[offset..][0..4]);
407 }
408
409 fn get64(input: *const [schema.stream_bytes]u8, offset: usize) u64 {
410 std.debug.assert(offset + 8 <= input.len);
411 return wire.read64(input[offset..][0..8]);
412 }