1	import Base
2	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
3	
4	# Reports whether the requested snapshot fits the retained-resource limits.
5	type Decision is Data:
6	  Retained{snapshot_count: Nat, retained_bytes: Nat}
7	  ResourceLimit{}
8	
9	# Holds the aggregate reservation counters for one database owner.
10	type Registry is Data:
11	  Registry{snapshot_count: Nat, retained_bytes: Nat}
12	
13	# Returns the updated registry or its unchanged prior state after rejection.
14	type RegistryDecision is Data:
15	  Admitted{registry: Registry}
16	  Rejected{registry: Registry}
17	
18	# Reports a snapshot release or preserves the registry on invalid release.
19	type RegistryReleaseDecision is Data:
20	  Released{registry: Registry}
21	  InvalidRelease{registry: Registry}
22	
23	# Couples snapshot closure to child checks and aggregate reservation release.
24	type SnapshotCloseDecision is Data:
25	  SnapshotClosed{registry: Registry}
26	  SnapshotBusy{registry: Registry}
27	  InvalidSnapshotClose{registry: Registry}
28	
29	# Tracks whether a snapshot is open and how many cursors still reference it.
30	type Lifecycle is Data:
31	  Live{cursor_count: Nat}
32	  Closed{}
33	
34	# Reports a lifecycle transition, a live-child conflict, or an invalid action.
35	type LifecycleDecision is Data:
36	  Advanced{lifecycle: Lifecycle}
37	  Busy{}
38	  Invalid{}
39	
40	# Reports a published generation and whether its predecessor remains retained.
41	type GenerationDecision is Data:
42	  Published{current: Nat, retired: RetiredGeneration}
43	
44	# Represents one older SST generation while snapshots still reference it.
45	type RetiredGeneration is Data:
46	  NoRetained{}
47	  GenerationRetained{generation: Nat, bytes: Nat, references: Nat}
48	
49	# Reports whether releasing a generation reference keeps or reclaims its files.
50	type RetentionDecision is Data:
51	  Keep{generation: Nat, bytes: Nat, references: Nat}
52	  Reclaim{generation: Nat, bytes: Nat}
53	  InvalidReference{}
54	
55	# Tracks one published SST generation and its live reader references.
56	type FileGeneration is Data:
57	  FileGeneration{id: Nat, size: Nat, references: Nat}
58	
59	# Keeps the current generation separate from all reader-retained generations.
60	type GenerationRegistry is Data:
61	  GenerationRegistry{current: FileGeneration,
62	    retired: List<&2, FileGeneration>}
63	
64	# Reports a successful generation pin or preserves the registry on a missing identifier.
65	type GenerationPinDecision is Data:
66	  Pinned{registry: GenerationRegistry}
67	  MissingGeneration{registry: GenerationRegistry}
68	
69	# Reports an incremented reader count for a located retired generation.
70	type RetiredPin is Data:
71	  RetiredPinned{retired: List<&2, FileGeneration>}
72	  RetiredPinMissing{retired: List<&2, FileGeneration>}
73	
74	# Reports a retained, reclaimed, or missing retired generation during release.
75	type RetiredRelease is Data:
76	  RetiredReleased{retired: List<&2, FileGeneration>}
77	  RetiredReclaimed{retired: List<&2, FileGeneration>, number: Nat, bytes: Nat}
78	  RetiredMissing{retired: List<&2, FileGeneration>}
79	
80	# Reports a decremented reference or identifies files reclaimed after the last reader.
81	type GenerationReleaseDecision is Data:
82	  GenerationReleased{registry: GenerationRegistry}
83	  GenerationReclaimed{registry: GenerationRegistry, number: Nat, bytes: Nat}
84	  MissingReference{registry: GenerationRegistry}
85	
86	# Carries the next owner resource identifier and terminal exhaustion state.
87	type ResourceIdAllocator is Data:
88	  ResourceIdAllocator{next_id: U32, exhausted: Bool}
89	
90	# Reports an allocated owner identifier or preserves an exhausted allocator.
91	type ResourceIdDecision is Data:
92	  ResourceIdAllocated{id: U32, allocator: ResourceIdAllocator}
93	  ResourceIdExhausted{allocator: ResourceIdAllocator}
94	
95	# Associates each owner-managed snapshot identifier with its generation and children.
96	type SnapshotResource is Data:
97	  SnapshotResource{id: U32, generation: Nat, retained_bytes: Nat,
98	    cursor_count: Nat}
99	
100	# Identifies one owner-managed cursor and its parent snapshot.
101	type CursorResource is Data:
102	  CursorResource{id: U32, snapshot_id: U32}
103	
104	# Holds aggregate snapshot reservations and their owner-issued identifiers.
105	type OwnerRegistry is Data:
106	  OwnerRegistry{allocator: ResourceIdAllocator, reservations: Registry,
107	    snapshots: List<&2, SnapshotResource>, cursors: List<&2, CursorResource>}
108	
109	# Reports an admitted owner snapshot or the exact registry after rejection.
110	type OwnerOpenDecision is Data:
111	  OwnerSnapshotOpened{registry: OwnerRegistry, snapshot_id: U32}
112	  OwnerSnapshotRejected{registry: OwnerRegistry}
113	
114	# Reports a cursor acquisition or the unchanged owner registry on rejection.
115	type OwnerCursorOpenDecision is Data:
116	  OwnerCursorOpened{registry: OwnerRegistry, cursor_id: U32}
117	  OwnerCursorRejected{registry: OwnerRegistry}
118	
119	# Reports cursor removal with enough context to restore it after invalid release.
120	type CursorListRemoveDecision is Data:
121	  CursorListRemoved{prefix: List<&2, CursorResource>,
122	    resource: CursorResource,
123	    tail: List<&2, CursorResource>}
124	  CursorListMissing{cursors: List<&2, CursorResource>}
125	
126	# Carries the recursive cursor-ID search with its untouched list prefix.
127	type CursorListRemoveState is Type:
128	  CursorSearch{cursors: List<&2, CursorResource>, cursor_id: U32,
129	    prefix: List<&2, CursorResource>}
130	  CursorCompareCopies{copies: CursorResource & CursorResource,
131	    tail: List<&2, CursorResource>, cursor_ids: U32 & U32,
132	    prefix: List<&2, CursorResource>}
133	  CursorCompare{matches: Bool, resource: CursorResource,
134	    tail: List<&2, CursorResource>, cursor_id: U32,
135	    prefix: List<&2, CursorResource>}
136	
137	# Reports cursor closure or preserves owner accounting on invalid release.
138	type OwnerCursorCloseDecision is Data:
139	  OwnerCursorClosed{registry: OwnerRegistry, snapshot_id: U32}
140	  OwnerCursorCloseRejected{registry: OwnerRegistry}
141	
142	# Reports snapshot closure, live-child Busy, or unchanged registry rejection.
143	type OwnerSnapshotCloseDecision is Data:
144	  OwnerSnapshotClosed{registry: OwnerRegistry, resource: SnapshotResource}
145	  OwnerSnapshotBusy{registry: OwnerRegistry}
146	  OwnerSnapshotCloseRejected{registry: OwnerRegistry}
147	
148	# Reports replacement of one snapshot's cursor count in the owner registry.
149	type SnapshotListCursorDecision is Data:
150	  SnapshotListUpdated{snapshots: List<&2, SnapshotResource>}
151	  SnapshotListRejected{snapshots: List<&2, SnapshotResource>}
152	  SnapshotListClosed{prefix: List<&2, SnapshotResource>,
153	    resource: SnapshotResource, tail: List<&2, SnapshotResource>}
154	  SnapshotListBusy{snapshots: List<&2, SnapshotResource>}
155	
156	# Selects whether a snapshot child reference is acquired or released.
157	type SnapshotCursorAction is Data:
158	  SnapshotCursorAcquire{max_cursors: Nat}
159	  SnapshotCursorRelease{}
160	  SnapshotCursorClose{}
161	
162	# Carries the recursive snapshot lookup without losing the original list prefix.
163	type SnapshotListCursorState is Type:
164	  SnapshotCursorSearch{snapshots: List<&2, SnapshotResource>,
165	    snapshot_id: U32, action: SnapshotCursorAction,
166	    prefix: List<&2, SnapshotResource>}
167	  SnapshotCursorCompareCopies{copies: SnapshotResource & SnapshotResource,
168	    tail: List<&2, SnapshotResource>, snapshot_ids: U32 & U32,
169	    action: SnapshotCursorAction, prefix: List<&2, SnapshotResource>}
170	  SnapshotCursorCompare{matches: Bool, resource: SnapshotResource,
171	    tail: List<&2, SnapshotResource>, snapshot_id: U32,
172	    action: SnapshotCursorAction,
173	    prefix: List<&2, SnapshotResource>}
174	
175	# Reports a snapshot cursor-reference update or preserves its prior resource.
176	type SnapshotCursorDecision is Data:
177	  SnapshotCursorAcquired{resource: SnapshotResource}
178	  SnapshotCursorLimit{resource: SnapshotResource}
179	  SnapshotCursorReleased{resource: SnapshotResource}
180	  SnapshotCursorUnderflow{resource: SnapshotResource}
181	
182	def point_read.matches(order: Cmp) -> Bool:
183	  match order:
184	    case EQ{}: True{}
185	    case _: False{}
186	
187	
188	
189	def nat.duplicate.go(value: Nat, first: Nat, second: Nat) -> Nat & Nat:
190	  match value:
191	    case 0n: (first, second)
192	    case 1n+rest:
193	      nat.duplicate.go(rest, Nat.add(1n, first), Nat.add(1n, second))
194	
195	def nat.duplicate(value: Nat) -> Nat & Nat:
196	  nat.duplicate.go(value, 0n, 0n)
197	
198	def allocate_resource_id.maximum(
199	  next_ids: U32 & U32,
200	  at_max: Bool
201	) -> ResourceIdDecision:
202	  match next_ids:
203	    case (next_id, increment_id):
204	      match at_max:
205	        case True{}:
206	          ResourceIdAllocated{next_id,
207	            ResourceIdAllocator{increment_id, True{}}}
208	        case False{}:
209	          ResourceIdAllocated{next_id,
210	            ResourceIdAllocator{U32.add(increment_id, 1), False{}}}
211	
212	def u32.bytes(value: U32) -> Bytes.Bytes:
213	  Bytes.set.u32be(Bytes.new(4), 0, value)
214	
215	def u32.clone.pair.results(
216	  results: (Bytes.Bytes & Maybe<&2, U32>) & (Bytes.Bytes & Maybe<&2, U32>)
217	) -> U32 & U32:
218	  match results:
219	    case ((_, Some{first}), (_, Some{second})): (first, second)
220	    case _: (0, 0)
221	
222	def u32.clone.pair.bytes(bytes: Bytes.Bytes & Bytes.Bytes) -> U32 & U32:
223	  match bytes:
224	    case (left, right):
225	      u32.clone.pair.results((Bytes.get.u32be(left, 0),
226	        Bytes.get.u32be(right, 0)))
227	
228	def u32.clone.pair.buffers(
229	  +len: U32,
230	  buffers: Array<U32> & Array<U32>
231	) -> U32 & U32:
232	  match buffers:
233	    case (left, right):
234	      u32.clone.pair.bytes((Bytes.Bytes{len, left}, Bytes.Bytes{len, right}))
235	
236	def u32.duplicate.encoded(bytes: Bytes.Bytes) -> U32 & U32:
237	  match bytes:
238	    case Bytes.Bytes{+len, buf}:
239	      u32.clone.pair.buffers(len,
240	        Bytes.copy.bytes(len, buf, Bytes.alloc(len), 0, 0))
241	
242	def u32.duplicate(value: U32) -> U32 & U32:
243	  u32.duplicate.encoded(u32.bytes(value))
244	
245	def allocate_resource_id.at_limit(
246	  copies: U32 & U32,
247	  exhausted: Bool
248	) -> ResourceIdDecision:
249	  match copies:
250	    case (next_id, check_id):
251	      match exhausted:
252	        case True{}:
253	          ResourceIdExhausted{ResourceIdAllocator{next_id, exhausted}}
254	        case False{}:
255	          allocate_resource_id.maximum(u32.duplicate(next_id),
256	            U32.is_eq(check_id, 4294967295))
257	
258	# Allocates one nonrecycled owner resource identifier without integer wraparound.
259	def allocate_resource_id(
260	  allocator: ResourceIdAllocator
261	) -> ResourceIdDecision:
262	  match allocator:
263	    case ResourceIdAllocator{next_id, exhausted}:
264	      allocate_resource_id.at_limit(u32.duplicate(next_id), exhausted)
265	
266	# Produces the updated reservation only when both bounds admit it.
267	def reserve.result(
268	  count: Nat,
269	  retained_bytes: Nat,
270	  admitted: Bool & Bool
271	) -> Decision:
272	  match admitted:
273	    case (True{}, True{}):
274	      Retained{Nat.add(count, 1n), retained_bytes}
275	    case _:
276	      ResourceLimit{}
277	
278	# Computes the updated retained-byte total and both admission checks.
279	def reserve.bytes(
280	  retained: Nat & Nat,
281	  requested: Nat & Nat,
282	  snapshot_count: Nat & Nat,
283	  max_snapshots: Nat,
284	  max_retained_bytes: Nat
285	) -> Nat & (Nat & (Bool & Bool)):
286	  match retained:
287	    case (retained_value, retained_check):
288	      match requested:
289	        case (requested_value, requested_check):
290	          match snapshot_count:
291	            case (count_output, count_check):
292	              (count_output,
293	                (Nat.add(retained_value, requested_value),
294	                  (Nat.is_lt(count_check, max_snapshots),
295	                  Nat.is_le(Nat.add(retained_check, requested_check), max_retained_bytes))))
296	
297	# Unpacks the reservation calculation into its public decision.
298	def reserve.result.checked(pair: Nat & (Nat & (Bool & Bool))) -> Decision:
299	  match pair:
300	    case (count, (retained_bytes, admitted)):
301	      reserve.result(count, retained_bytes, admitted)
302	
303	# Decides whether one snapshot fits the configured count and byte limits.
304	def reserve(
305	  snapshot_count: Nat,
306	  retained_bytes: Nat,
307	  requested_bytes: Nat,
308	  max_snapshots: Nat,
309	  max_retained_bytes: Nat
310	) -> Decision:
311	  reserve.result.checked(reserve.bytes(nat.duplicate(retained_bytes),
312	    nat.duplicate(requested_bytes), nat.duplicate(snapshot_count), max_snapshots,
313	    max_retained_bytes))
314	
315	# Preserves the prior registry when a reservation fails either configured bound.
316	def registry.reserve.result(original: Registry, decision: Decision) -> RegistryDecision:
317	  match decision:
318	    case Retained{snapshot_count, retained_bytes}:
319	      Admitted{Registry{snapshot_count, retained_bytes}}
320	    case ResourceLimit{}: Rejected{original}
321	
322	# Applies a reservation decision while keeping a copy of the prior counters.
323	def registry.reserve.input(
324	  copies: (Nat & Nat) & (Nat & Nat),
325	  requested_bytes: Nat,
326	  max_snapshots: Nat,
327	  max_retained_bytes: Nat
328	) -> RegistryDecision:
329	  match copies:
330	    case ((count_input, count_original), (bytes_input, bytes_original)):
331	      registry.reserve.result(Registry{count_original, bytes_original},
332	      reserve(count_input, bytes_input, requested_bytes,
333	          max_snapshots, max_retained_bytes))
334	
335	# Duplicates the byte counter so an admitted decision can update while rejection preserves it.
336	def registry.reserve.count(
337	  count_copies: Nat & Nat,
338	  retained_bytes: Nat,
339	  requested_bytes: Nat,
340	  max_snapshots: Nat,
341	  max_retained_bytes: Nat
342	) -> RegistryDecision:
343	  match count_copies:
344	    case (count_input, count_original):
345	      registry.reserve.input(((count_input, count_original),
346	        nat.duplicate(retained_bytes)), requested_bytes,
347	        max_snapshots, max_retained_bytes)
348	
349	# Reserves one snapshot against the aggregate counters and preserves state on rejection.
350	def registry.reserve(
351	  registry: Registry,
352	  requested_bytes: Nat,
353	  max_snapshots: Nat,
354	  max_retained_bytes: Nat
355	) -> RegistryDecision:
356	  match registry:
357	    case Registry{snapshot_count, retained_bytes}:
358	      registry.reserve.count(nat.duplicate(snapshot_count), retained_bytes,
359	        requested_bytes, max_snapshots, max_retained_bytes)
360	
361	# Keeps the original counters when release preconditions do not hold.
362	def registry.release.result(
363	  counters: Nat & Nat,
364	  original: Registry,
365	  admitted: Bool & Bool
366	) -> RegistryReleaseDecision:
367	  match counters:
368	    case (snapshot_count, retained_bytes):
369	      match admitted:
370	        case (True{}, True{}):
371	          Released{Registry{snapshot_count, retained_bytes}}
372	        case _: InvalidRelease{original}
373	
374	# Separates byte arithmetic inputs so both admission checks retain their values.
375	def registry.release.input.bytes(
376	  count_copies: Nat & Nat,
377	  count_original: Nat,
378	  bytes_copies: Nat & Nat,
379	  bytes_original: Nat,
380	  release_copies: Nat & Nat
381	) -> RegistryReleaseDecision:
382	  match count_copies:
383	    case (count_for_subtract, count_for_check):
384	      match bytes_copies:
385	        case (bytes_for_subtract, bytes_for_check):
386	          match release_copies:
387	            case (released_for_subtract, released_for_check):
388	              registry.release.result(
389	                (Nat.sub(count_for_subtract, 1n),
390	                  Nat.sub(bytes_for_subtract, released_for_subtract)),
391	                Registry{count_original, bytes_original},
392	                (Nat.is_lt(0n, count_for_check),
393	                  Nat.is_le(released_for_check, bytes_for_check)))
394	
395	# Validates positive snapshot ownership and sufficient retained bytes before decrementing.
396	def registry.release.input(
397	  copies: (Nat & Nat) & (Nat & Nat),
398	  released_bytes: Nat
399	) -> RegistryReleaseDecision:
400	  match copies:
401	    case ((count_input, count_original), (bytes_input, bytes_original)):
402	      registry.release.input.bytes(nat.duplicate(count_input), count_original,
403	        nat.duplicate(bytes_input), bytes_original,
404	        nat.duplicate(released_bytes))
405	
406	# Duplicates the counters so invalid releases preserve the exact prior registry.
407	def registry.release.count(
408	  count_copies: Nat & Nat,
409	  retained_bytes: Nat,
410	  released_bytes: Nat
411	) -> RegistryReleaseDecision:
412	  match count_copies:
413	    case (count_input, count_original):
414	      registry.release.input(((count_input, count_original),
415	        nat.duplicate(retained_bytes)), released_bytes)
416	
417	# Releases one closed snapshot's reservation without changing active readers.
418	def registry.release(registry: Registry, released_bytes: Nat) -> RegistryReleaseDecision:
419	  match registry:
420	    case Registry{snapshot_count, retained_bytes}:
421	      registry.release.count(nat.duplicate(snapshot_count), retained_bytes,
422	        released_bytes)
423	
424	def owner.snapshot.open.release(
425	  allocator: ResourceIdAllocator,
426	  snapshots: List<&2, SnapshotResource>,
427	  cursors: List<&2, CursorResource>,
428	  decision: RegistryReleaseDecision
429	) -> OwnerOpenDecision:
430	  match decision:
431	    case Released{reservations}:
432	      OwnerSnapshotRejected{OwnerRegistry{allocator, reservations, snapshots,
433	        cursors}}
434	    case InvalidRelease{reservations}:
435	      OwnerSnapshotRejected{OwnerRegistry{allocator, reservations, snapshots,
436	        cursors}}
437	
438	def owner.snapshot.open.created(
439	  ids: U32 & U32,
440	  allocator: ResourceIdAllocator,
441	  reservations: Registry,
442	  snapshots: List<&2, SnapshotResource>,
443	  cursors: List<&2, CursorResource>,
444	  generation: Nat,
445	  retained_bytes: Nat
446	) -> OwnerOpenDecision:
447	  match ids:
448	    case (snapshot_id, entry_id):
449	      OwnerSnapshotOpened{OwnerRegistry{allocator, reservations,
450	        Con{SnapshotResource{entry_id, generation, retained_bytes, 0n}, snapshots},
451	        cursors},
452	        snapshot_id}
453	
454	def owner.snapshot.open.id(
455	  allocation: ResourceIdDecision,
456	  reservations: Registry,
457	  snapshots: List<&2, SnapshotResource>,
458	  cursors: List<&2, CursorResource>,
459	  generation: Nat,
460	  retained_bytes: Nat
461	) -> OwnerOpenDecision:
462	  match allocation:
463	    case ResourceIdAllocated{id, allocator}:
464	      owner.snapshot.open.created(u32.duplicate(id), allocator, reservations,
465	        snapshots, cursors, generation, retained_bytes)
466	    case ResourceIdExhausted{allocator}:
467	      owner.snapshot.open.release(allocator, snapshots, cursors,
468	        registry.release(reservations, retained_bytes))
469	
470	def owner.snapshot.open.reserved(
471	  allocator: ResourceIdAllocator,
472	  snapshots: List<&2, SnapshotResource>,
473	  cursors: List<&2, CursorResource>,
474	  generation: Nat,
475	  requests: Nat & Nat,
476	  decision: RegistryDecision
477	) -> OwnerOpenDecision:
478	  match requests:
479	    case (retained_bytes, release_bytes):
480	      match decision:
481	        case Admitted{reservations}:
482	          owner.snapshot.open.id(allocate_resource_id(allocator), reservations,
483	            snapshots, cursors, generation, release_bytes)
484	        case Rejected{reservations}:
485	          OwnerSnapshotRejected{OwnerRegistry{allocator, reservations,
486	            snapshots, cursors}}
487	
488	def owner.snapshot.open.budget(
489	  allocator: ResourceIdAllocator,
490	  snapshots: List<&2, SnapshotResource>,
491	  cursors: List<&2, CursorResource>,
492	  generation: Nat,
493	  requested_bytes: Nat & Nat,
494	  release_bytes: Nat,
495	  limits: Nat & Nat,
496	  reservations: Registry
497	) -> OwnerOpenDecision:
498	  match requested_bytes:
499	    case (retained_for_entry, retained_for_admission):
500	      match limits:
501	        case (max_snapshots, max_retained_bytes):
502	          owner.snapshot.open.reserved(allocator, snapshots, cursors,
503	            generation, (retained_for_entry, release_bytes),
504	            registry.reserve(reservations, retained_for_admission,
505	              max_snapshots, max_retained_bytes))
506	
507	def owner.snapshot.open.available(
508	  allocator: ResourceIdAllocator,
509	  reservations: Registry,
510	  snapshots: List<&2, SnapshotResource>,
511	  cursors: List<&2, CursorResource>,
512	  generation: Nat,
513	  requests: Nat & Nat,
514	  limits: Nat & Nat
515	) -> OwnerOpenDecision:
516	  match allocator:
517	    case ResourceIdAllocator{next_id, exhausted}:
518	      match exhausted:
519	        case True{}:
520	          OwnerSnapshotRejected{OwnerRegistry{allocator, reservations,
521	            snapshots, cursors}}
522	        case False{}:
523	          match requests:
524	            case (requested_bytes, release_bytes):
525	              owner.snapshot.open.budget(allocator, snapshots, cursors,
526	                generation, nat.duplicate(requested_bytes), release_bytes, limits,
527	                reservations)
528	
529	def owner.snapshot.open(
530	  owner: OwnerRegistry,
531	  generation: Nat,
532	  requested_bytes: Nat,
533	  limits: Nat & Nat
534	) -> OwnerOpenDecision:
535	  match owner:
536	    case OwnerRegistry{allocator, reservations, snapshots, cursors}:
537	      owner.snapshot.open.available(allocator, reservations, snapshots,
538	        cursors,
539	        generation, nat.duplicate(requested_bytes), limits)
540	
541	def snapshot.resource.duplicate.parts(
542	  ids: U32 & U32,
543	  generations: Nat & Nat,
544	  retained: Nat & Nat,
545	  cursors: Nat & Nat
546	) -> SnapshotResource & SnapshotResource:
547	  match ids:
548	    case (first_id, second_id):
549	      match generations:
550	        case (first_generation, second_generation):
551	          match retained:
552	            case (first_bytes, second_bytes):
553	              match cursors:
554	                case (first_count, second_count):
555	                  (SnapshotResource{first_id, first_generation, first_bytes,
556	                    first_count}, SnapshotResource{second_id, second_generation,
557	                    second_bytes, second_count})
558	
559	def snapshot.resource.duplicate(resource: SnapshotResource) -> SnapshotResource & SnapshotResource:
560	  match resource:
561	    case SnapshotResource{id, generation, retained_bytes, cursor_count}:
562	      snapshot.resource.duplicate.parts(u32.duplicate(id),
563	        nat.duplicate(generation), nat.duplicate(retained_bytes),
564	        nat.duplicate(cursor_count))
565	
566	def snapshot.resource.cursor.acquire.result(
567	  resource: SnapshotResource,
568	  admitted: Bool
569	) -> SnapshotCursorDecision:
570	  match resource:
571	    case SnapshotResource{id, generation, retained_bytes, cursor_count}:
572	      match admitted:
573	        case True{}:
574	          SnapshotCursorAcquired{SnapshotResource{id, generation,
575	            retained_bytes, Nat.add(cursor_count, 1n)}}
576	        case False{}: SnapshotCursorLimit{resource}
577	
578	def snapshot.resource.cursor.acquire.checked(
579	  resource: SnapshotResource,
580	  cursor_count: Nat,
581	  max_cursors: Nat
582	) -> SnapshotCursorDecision:
583	  snapshot.resource.cursor.acquire.result(resource,
584	    Nat.is_lt(cursor_count, max_cursors))
585	
586	def snapshot.resource.cursor.acquire.copied(
587	  copies: SnapshotResource & SnapshotResource,
588	  max_cursors: Nat
589	) -> SnapshotCursorDecision:
590	  match copies:
591	    case (for_check, for_update):
592	      match for_check:
593	        case SnapshotResource{_, _, _, cursor_count}:
594	          snapshot.resource.cursor.acquire.checked(for_update, cursor_count,
595	            max_cursors)
596	
597	# Acquires one child reference without changing a snapshot at its cursor limit.
598	def snapshot.resource.cursor.acquire(
599	  resource: SnapshotResource,
600	  max_cursors: Nat
601	) -> SnapshotCursorDecision:
602	  snapshot.resource.cursor.acquire.copied(
603	    snapshot.resource.duplicate(resource), max_cursors)
604	
605	def snapshot.resource.cursor.release.result(
606	  resource: SnapshotResource,
607	  has_reference: Bool
608	) -> SnapshotCursorDecision:
609	  match resource:
610	    case SnapshotResource{id, generation, retained_bytes, cursor_count}:
611	      match has_reference:
612	        case True{}:
613	          SnapshotCursorReleased{SnapshotResource{id, generation,
614	            retained_bytes, Nat.sub(cursor_count, 1n)}}
615	        case False{}: SnapshotCursorUnderflow{resource}
616	
617	def snapshot.resource.cursor.release.checked(
618	  resource: SnapshotResource,
619	  cursor_count: Nat
620	) -> SnapshotCursorDecision:
621	  snapshot.resource.cursor.release.result(resource,
622	    Nat.is_lt(0n, cursor_count))
623	
624	def snapshot.resource.cursor.release.copied(
625	  copies: SnapshotResource & SnapshotResource
626	) -> SnapshotCursorDecision:
627	  match copies:
628	    case (for_check, for_update):
629	      match for_check:
630	        case SnapshotResource{_, _, _, cursor_count}:
631	          snapshot.resource.cursor.release.checked(for_update, cursor_count)
632	
633	# Releases one child reference and preserves the resource on underflow.
634	def snapshot.resource.cursor.release(
635	  resource: SnapshotResource
636	) -> SnapshotCursorDecision:
637	  snapshot.resource.cursor.release.copied(
638	    snapshot.resource.duplicate(resource))
639	
640	def snapshot.list.cursor.restore(
641	  prefix: List<&2, SnapshotResource>,
642	  snapshots: List<&2, SnapshotResource>
643	) -> List<&2, SnapshotResource>:
644	  match prefix:
645	    case Nil{}: snapshots
646	    case Con{resource, tail}:
647	      snapshot.list.cursor.restore(tail, Con{resource, snapshots})
648	
649	def snapshot.list.cursor.length.keep.go(
650	  snapshots: List<&2, SnapshotResource>,
651	  reversed: List<&2, SnapshotResource>,
652	  length: Nat
653	) -> List<&2, SnapshotResource> & Nat:
654	  match snapshots:
655	    case Nil{}:
656	      (List.reverse(&2, SnapshotResource, reversed), length)
657	    case Con{resource, tail}:
658	      snapshot.list.cursor.length.keep.go(tail, Con{resource, reversed},
659	        Nat.add(length, 1n))
660	
661	def snapshot.list.cursor.length.keep(
662	  snapshots: List<&2, SnapshotResource>
663	) -> List<&2, SnapshotResource> & Nat:
664	  snapshot.list.cursor.length.keep.go(snapshots, Nil{}, 0n)
665	
666	def snapshot.list.cursor.resolve(
667	  prefix: List<&2, SnapshotResource>,
668	  tail: List<&2, SnapshotResource>,
669	  acquired: SnapshotCursorDecision
670	) -> SnapshotListCursorDecision:
671	  match acquired:
672	    case SnapshotCursorAcquired{resource}:
673	      SnapshotListUpdated{snapshot.list.cursor.restore(prefix,
674	        Con{resource, tail})}
675	    case SnapshotCursorLimit{resource}:
676	      SnapshotListRejected{snapshot.list.cursor.restore(prefix,
677	        Con{resource, tail})}
678	    case SnapshotCursorReleased{resource}:
679	      SnapshotListUpdated{snapshot.list.cursor.restore(prefix,
680	        Con{resource, tail})}
681	    case SnapshotCursorUnderflow{resource}:
682	      SnapshotListRejected{snapshot.list.cursor.restore(prefix,
683	        Con{resource, tail})}
684	
685	def snapshot.list.cursor.close.result(
686	  can_close: Bool,
687	  prefix: List<&2, SnapshotResource>,
688	  resource: SnapshotResource,
689	  tail: List<&2, SnapshotResource>
690	) -> SnapshotListCursorDecision:
691	  match can_close:
692	    case True{}: SnapshotListClosed{prefix, resource, tail}
693	    case False{}:
694	      SnapshotListBusy{snapshot.list.cursor.restore(prefix,
695	        Con{resource, tail})}
696	
697	def snapshot.list.cursor.close.resource(
698	  copies: SnapshotResource & SnapshotResource,
699	  prefix: List<&2, SnapshotResource>,
700	  tail: List<&2, SnapshotResource>
701	) -> SnapshotListCursorDecision:
702	  match copies:
703	    case (for_check, for_update):
704	      match for_check:
705	        case SnapshotResource{_, _, _, cursor_count}:
706	          snapshot.list.cursor.close.result(Nat.is_eq(cursor_count, 0n),
707	            prefix, for_update, tail)
708	
709	def snapshot.list.cursor.action(
710	  prefix: List<&2, SnapshotResource>,
711	  tail: List<&2, SnapshotResource>,
712	  resource: SnapshotResource,
713	  action: SnapshotCursorAction
714	) -> SnapshotListCursorDecision:
715	  match action:
716	    case SnapshotCursorAcquire{max_cursors}:
717	      snapshot.list.cursor.resolve(prefix, tail,
718	        snapshot.resource.cursor.acquire(resource, max_cursors))
719	    case SnapshotCursorRelease{}:
720	      snapshot.list.cursor.resolve(prefix, tail,
721	        snapshot.resource.cursor.release(resource))
722	    case SnapshotCursorClose{}:
723	      snapshot.list.cursor.close.resource(
724	        snapshot.resource.duplicate(resource), prefix, tail)
725	
726	def snapshot.list.cursor.exhausted(
727	  state: SnapshotListCursorState
728	) -> SnapshotListCursorDecision:
729	  match state:
730	    case SnapshotCursorSearch{snapshots, _, _, prefix}:
731	      SnapshotListRejected{snapshot.list.cursor.restore(prefix, snapshots)}
732	    case SnapshotCursorCompareCopies{copies, tail, _, _, prefix}:
733	      match copies:
734	        case (_, resource):
735	          SnapshotListRejected{snapshot.list.cursor.restore(prefix,
736	            Con{resource, tail})}
737	    case SnapshotCursorCompare{_, resource, tail, _, _, prefix}:
738	      SnapshotListRejected{snapshot.list.cursor.restore(prefix,
739	        Con{resource, tail})}
740	
741	# Updates exactly one snapshot cursor count and preserves every unrelated entry.
742	def snapshot.list.cursor.acquire.loop(
743	  fuel: Nat,
744	  state: SnapshotListCursorState
745	) -> SnapshotListCursorDecision:
746	  match fuel:
747	    case 0n:
748	      snapshot.list.cursor.exhausted(state)
749	    case 1n+rest:
750	      match state:
751	        case SnapshotCursorSearch{snapshots, snapshot_id, action,
752	            prefix}:
753	          match snapshots:
754	            case Nil{}:
755	              SnapshotListRejected{snapshot.list.cursor.restore(prefix,
756	                Nil{})}
757	            case Con{resource, tail}:
758	              snapshot.list.cursor.acquire.loop(rest,
759	                SnapshotCursorCompareCopies{
760	                  snapshot.resource.duplicate(resource), tail,
761	                  u32.duplicate(snapshot_id), action, prefix})
762	        case SnapshotCursorCompareCopies{copies, tail, snapshot_ids,
763	            action, prefix}:
764	          match copies:
765	            case (for_check, for_update):
766	              match for_check:
767	                case SnapshotResource{id, _, _, _}:
768	                  match snapshot_ids:
769	                    case (current_id, next_id):
770	                      snapshot.list.cursor.acquire.loop(rest,
771	                        SnapshotCursorCompare{U32.is_eq(id, current_id),
772	                          for_update, tail, next_id, action, prefix})
773	        case SnapshotCursorCompare{matches, resource, tail, snapshot_id,
774	            action, prefix}:
775	          match matches:
776	            case True{}:
777	              snapshot.list.cursor.action(prefix, tail, resource, action)
778	            case False{}:
779	              snapshot.list.cursor.acquire.loop(rest,
780	                SnapshotCursorSearch{tail, snapshot_id, action,
781	                  Con{resource, prefix}})
782	
783	def snapshot.list.cursor.acquire.measured(
784	  measured: List<&2, SnapshotResource> & Nat,
785	  snapshot_id: U32,
786	  action: SnapshotCursorAction
787	) -> SnapshotListCursorDecision:
788	  match measured:
789	    case (snapshots, length):
790	      snapshot.list.cursor.acquire.loop(
791	        Nat.add(Nat.mul(3n, length), 1n),
792	        SnapshotCursorSearch{snapshots, snapshot_id, action, Nil{}})
793	
794	# Adds one cursor reference to the owner record identified by its snapshot ID.
795	def snapshot.list.cursor.acquire(
796	  snapshots: List<&2, SnapshotResource>,
797	  snapshot_id: U32,
798	  max_cursors: Nat
799	) -> SnapshotListCursorDecision:
800	  snapshot.list.cursor.acquire.measured(
801	    snapshot.list.cursor.length.keep(snapshots), snapshot_id,
802	    SnapshotCursorAcquire{max_cursors})
803	
804	# Releases exactly one cursor reference from the matching snapshot resource.
805	def snapshot.list.cursor.release(
806	  snapshots: List<&2, SnapshotResource>,
807	  snapshot_id: U32
808	) -> SnapshotListCursorDecision:
809	  snapshot.list.cursor.acquire.measured(
810	    snapshot.list.cursor.length.keep(snapshots), snapshot_id,
811	    SnapshotCursorRelease{})
812	
813	# Removes a childless snapshot record and reports Busy while cursors remain.
814	def snapshot.list.cursor.close(
815	  snapshots: List<&2, SnapshotResource>,
816	  snapshot_id: U32
817	) -> SnapshotListCursorDecision:
818	  snapshot.list.cursor.acquire.measured(
819	    snapshot.list.cursor.length.keep(snapshots), snapshot_id,
820	    SnapshotCursorClose{})
821	
822	def owner.cursor.open.ready(
823	  cursor_ids: U32 & U32,
824	  allocator: ResourceIdAllocator,
825	  reservations: Registry,
826	  snapshots: List<&2, SnapshotResource>,
827	  cursors: List<&2, CursorResource>,
828	  snapshot_id: U32
829	) -> OwnerCursorOpenDecision:
830	  match cursor_ids:
831	    case (cursor_id, entry_id):
832	      OwnerCursorOpened{OwnerRegistry{allocator, reservations, snapshots,
833	        Con{CursorResource{entry_id, snapshot_id}, cursors}}, cursor_id}
834	
835	def owner.cursor.open.allocated(
836	  allocation: ResourceIdDecision,
837	  reservations: Registry,
838	  snapshots: List<&2, SnapshotResource>,
839	  cursors: List<&2, CursorResource>,
840	  snapshot_id: U32
841	) -> OwnerCursorOpenDecision:
842	  match allocation:
843	    case ResourceIdAllocated{cursor_id, next_allocator}:
844	      owner.cursor.open.ready(u32.duplicate(cursor_id), next_allocator,
845	        reservations, snapshots, cursors, snapshot_id)
846	    case ResourceIdExhausted{allocator}:
847	      OwnerCursorRejected{OwnerRegistry{allocator, reservations,
848	        snapshots, cursors}}
849	
850	def owner.cursor.open.snapshot(
851	  allocator: ResourceIdAllocator,
852	  reservations: Registry,
853	  cursors: List<&2, CursorResource>,
854	  snapshot_id: U32,
855	  updated: SnapshotListCursorDecision
856	) -> OwnerCursorOpenDecision:
857	  match updated:
858	    case SnapshotListUpdated{snapshots}:
859	      owner.cursor.open.allocated(allocate_resource_id(allocator),
860	        reservations, snapshots, cursors, snapshot_id)
861	    case SnapshotListRejected{snapshots}:
862	      OwnerCursorRejected{OwnerRegistry{allocator, reservations, snapshots,
863	        cursors}}
864	    case SnapshotListClosed{prefix, resource, tail}:
865	      OwnerCursorRejected{OwnerRegistry{allocator, reservations,
866	        snapshot.list.cursor.restore(prefix, Con{resource, tail}), cursors}}
867	    case SnapshotListBusy{snapshots}:
868	      OwnerCursorRejected{OwnerRegistry{allocator, reservations, snapshots,
869	        cursors}}
870	
871	def owner.cursor.open.request(
872	  allocator: ResourceIdAllocator,
873	  reservations: Registry,
874	  snapshots: List<&2, SnapshotResource>,
875	  cursors: List<&2, CursorResource>,
876	  snapshot_ids: U32 & U32,
877	  max_cursors: Nat
878	) -> OwnerCursorOpenDecision:
879	  match snapshot_ids:
880	    case (cursor_snapshot_id, lookup_snapshot_id):
881	      owner.cursor.open.snapshot(allocator, reservations, cursors,
882	        cursor_snapshot_id,
883	        snapshot.list.cursor.acquire(snapshots, lookup_snapshot_id,
884	          max_cursors))
885	
886	# Rejects cursor opens when the owner ID space is exhausted.
887	def owner.cursor.open.available(
888	  owner: OwnerRegistry,
889	  snapshot_id: U32,
890	  max_cursors: Nat
891	) -> OwnerCursorOpenDecision:
892	  match owner:
893	    case OwnerRegistry{allocator, reservations, snapshots, cursors}:
894	      match allocator:
895	        case ResourceIdAllocator{_, exhausted}:
896	          match exhausted:
897	            case True{}:
898	              OwnerCursorRejected{OwnerRegistry{allocator, reservations,
899	                snapshots, cursors}}
900	            case False{}:
901	              owner.cursor.open.request(allocator, reservations, snapshots,
902	                cursors, u32.duplicate(snapshot_id), max_cursors)
903	
904	# Acquires a unique owner cursor ID and a child reference on its snapshot.
905	def owner.cursor.open(
906	  owner: OwnerRegistry,
907	  snapshot_id: U32,
908	  max_cursors: Nat
909	) -> OwnerCursorOpenDecision:
910	  owner.cursor.open.available(owner, snapshot_id, max_cursors)
911	
912	def cursor.resource.duplicate.parts(
913	  ids: U32 & U32,
914	  snapshots: U32 & U32
915	) -> CursorResource & CursorResource:
916	  match ids:
917	    case (first_id, second_id):
918	      match snapshots:
919	        case (first_snapshot, second_snapshot):
920	          (CursorResource{first_id, first_snapshot},
921	            CursorResource{second_id, second_snapshot})
922	
923	def cursor.resource.duplicate(resource: CursorResource) -> CursorResource & CursorResource:
924	  match resource:
925	    case CursorResource{id, snapshot_id}:
926	      cursor.resource.duplicate.parts(u32.duplicate(id),
927	        u32.duplicate(snapshot_id))
928	
929	def cursor.list.restore(
930	  prefix: List<&2, CursorResource>,
931	  cursors: List<&2, CursorResource>
932	) -> List<&2, CursorResource>:
933	  match prefix:
934	    case Nil{}: cursors
935	    case Con{resource, tail}:
936	      cursor.list.restore(tail, Con{resource, cursors})
937	
938	def cursor.list.length.keep.go(
939	  cursors: List<&2, CursorResource>,
940	  reversed: List<&2, CursorResource>,
941	  length: Nat
942	) -> List<&2, CursorResource> & Nat:
943	  match cursors:
944	    case Nil{}: (List.reverse(&2, CursorResource, reversed), length)
945	    case Con{resource, tail}:
946	      cursor.list.length.keep.go(tail, Con{resource, reversed},
947	        Nat.add(length, 1n))
948	
949	def cursor.list.length.keep(
950	  cursors: List<&2, CursorResource>
951	) -> List<&2, CursorResource> & Nat:
952	  cursor.list.length.keep.go(cursors, Nil{}, 0n)
953	
954	def cursor.list.remove.exhausted(
955	  state: CursorListRemoveState
956	) -> CursorListRemoveDecision:
957	  match state:
958	    case CursorSearch{cursors, _, prefix}:
959	      CursorListMissing{cursor.list.restore(prefix, cursors)}
960	    case CursorCompareCopies{copies, tail, _, prefix}:
961	      match copies:
962	        case (_, resource):
963	          CursorListMissing{cursor.list.restore(prefix,
964	            Con{resource, tail})}
965	    case CursorCompare{_, resource, tail, _, prefix}:
966	      CursorListMissing{cursor.list.restore(prefix, Con{resource, tail})}
967	
968	def cursor.list.remove.loop(
969	  fuel: Nat,
970	  state: CursorListRemoveState
971	) -> CursorListRemoveDecision:
972	  match fuel:
973	    case 0n: cursor.list.remove.exhausted(state)
974	    case 1n+rest:
975	      match state:
976	        case CursorSearch{cursors, cursor_id, prefix}:
977	          match cursors:
978	            case Nil{}:
979	              CursorListMissing{cursor.list.restore(prefix, Nil{})}
980	            case Con{resource, tail}:
981	              cursor.list.remove.loop(rest, CursorCompareCopies{
982	                cursor.resource.duplicate(resource), tail,
983	                u32.duplicate(cursor_id), prefix})
984	        case CursorCompareCopies{copies, tail, cursor_ids, prefix}:
985	          match copies:
986	            case (for_check, for_update):
987	              match for_check:
988	                case CursorResource{id, _}:
989	                  match cursor_ids:
990	                    case (current_id, next_id):
991	                      cursor.list.remove.loop(rest,
992	                        CursorCompare{U32.is_eq(id, current_id), for_update,
993	                          tail, next_id, prefix})
994	        case CursorCompare{matches, resource, tail, cursor_id, prefix}:
995	          match matches:
996	            case True{}:
997	              CursorListRemoved{prefix, resource, tail}
998	            case False{}:
999	              cursor.list.remove.loop(rest, CursorSearch{tail, cursor_id,
1000	                Con{resource, prefix}})
1001	
1002	def cursor.list.remove.measured(
1003	  measured: List<&2, CursorResource> & Nat,
1004	  cursor_id: U32
1005	) -> CursorListRemoveDecision:
1006	  match measured:
1007	    case (cursors, length):
1008	      cursor.list.remove.loop(Nat.add(Nat.mul(3n, length), 1n),
1009	        CursorSearch{cursors, cursor_id, Nil{}})
1010	
1011	# Removes one owner cursor ID without changing its parent snapshot count.
1012	def cursor.list.remove(
1013	  cursors: List<&2, CursorResource>,
1014	  cursor_id: U32
1015	) -> CursorListRemoveDecision:
1016	  cursor.list.remove.measured(cursor.list.length.keep(cursors), cursor_id)
1017	
1018	def owner.cursor.close.released(
1019	  allocator: ResourceIdAllocator,
1020	  reservations: Registry,
1021	  prefix: List<&2, CursorResource>,
1022	  resource: CursorResource,
1023	  tail: List<&2, CursorResource>,
1024	  snapshot_id: U32,
1025	  released: SnapshotListCursorDecision
1026	) -> OwnerCursorCloseDecision:
1027	  match released:
1028	    case SnapshotListUpdated{snapshots}:
1029	      OwnerCursorClosed{OwnerRegistry{allocator, reservations, snapshots,
1030	        cursor.list.restore(prefix, tail)}, snapshot_id}
1031	    case SnapshotListRejected{snapshots}:
1032	      OwnerCursorCloseRejected{OwnerRegistry{allocator, reservations,
1033	        snapshots, cursor.list.restore(prefix, Con{resource, tail})}}
1034	    case SnapshotListClosed{snapshot_prefix, snapshot_resource,
1035	        snapshot_tail}:
1036	      OwnerCursorCloseRejected{OwnerRegistry{allocator, reservations,
1037	        snapshot.list.cursor.restore(snapshot_prefix,
1038	          Con{snapshot_resource, snapshot_tail}),
1039	        cursor.list.restore(prefix, Con{resource, tail})}}
1040	    case SnapshotListBusy{snapshots}:
1041	      OwnerCursorCloseRejected{OwnerRegistry{allocator, reservations,
1042	        snapshots, cursor.list.restore(prefix, Con{resource, tail})}}
1043	
1044	def owner.cursor.close.release.id(
1045	  allocator: ResourceIdAllocator,
1046	  reservations: Registry,
1047	  snapshots: List<&2, SnapshotResource>,
1048	  prefix: List<&2, CursorResource>,
1049	  resource: CursorResource,
1050	  tail: List<&2, CursorResource>,
1051	  snapshot_ids: U32 & U32
1052	) -> OwnerCursorCloseDecision:
1053	  match snapshot_ids:
1054	    case (close_snapshot_id, lookup_snapshot_id):
1055	      owner.cursor.close.released(allocator, reservations, prefix, resource,
1056	        tail, close_snapshot_id,
1057	        snapshot.list.cursor.release(snapshots, lookup_snapshot_id))
1058	
1059	def owner.cursor.close.lookup(
1060	  allocator: ResourceIdAllocator,
1061	  reservations: Registry,
1062	  snapshots: List<&2, SnapshotResource>,
1063	  prefix: List<&2, CursorResource>,
1064	  tail: List<&2, CursorResource>,
1065	  copies: CursorResource & CursorResource
1066	) -> OwnerCursorCloseDecision:
1067	  match copies:
1068	    case (for_check, for_update):
1069	      match for_check:
1070	        case CursorResource{_, snapshot_id}:
1071	          owner.cursor.close.release.id(allocator, reservations, snapshots,
1072	            prefix, for_update, tail, u32.duplicate(snapshot_id))
1073	
1074	def owner.cursor.close.removed(
1075	  allocator: ResourceIdAllocator,
1076	  reservations: Registry,
1077	  snapshots: List<&2, SnapshotResource>,
1078	  removal: CursorListRemoveDecision
1079	) -> OwnerCursorCloseDecision:
1080	  match removal:
1081	    case CursorListMissing{cursors}:
1082	      OwnerCursorCloseRejected{OwnerRegistry{allocator, reservations,
1083	        snapshots, cursors}}
1084	    case CursorListRemoved{prefix, resource, tail}:
1085	      owner.cursor.close.lookup(allocator, reservations, snapshots,
1086	        prefix, tail, cursor.resource.duplicate(resource))
1087	
1088	# Releases one owner cursor and its parent snapshot reference atomically.
1089	def owner.cursor.close(
1090	  owner: OwnerRegistry,
1091	  cursor_id: U32
1092	) -> OwnerCursorCloseDecision:
1093	  match owner:
1094	    case OwnerRegistry{allocator, reservations, snapshots, cursors}:
1095	      owner.cursor.close.removed(allocator, reservations, snapshots,
1096	        cursor.list.remove(cursors, cursor_id))
1097	
1098	def owner.snapshot.close.released(
1099	  decision: RegistryReleaseDecision,
1100	  allocator: ResourceIdAllocator,
1101	  cursors: List<&2, CursorResource>,
1102	  prefix: List<&2, SnapshotResource>,
1103	  resource: SnapshotResource,
1104	  tail: List<&2, SnapshotResource>
1105	) -> OwnerSnapshotCloseDecision:
1106	  match decision:
1107	    case Released{reservations}:
1108	      OwnerSnapshotClosed{OwnerRegistry{allocator, reservations,
1109	        snapshot.list.cursor.restore(prefix, tail), cursors}, resource}
1110	    case InvalidRelease{reservations}:
1111	      OwnerSnapshotCloseRejected{OwnerRegistry{allocator, reservations,
1112	        snapshot.list.cursor.restore(prefix, Con{resource, tail}), cursors}}
1113	
1114	def owner.snapshot.close.resource(
1115	  copies: SnapshotResource & SnapshotResource,
1116	  allocator: ResourceIdAllocator,
1117	  cursors: List<&2, CursorResource>,
1118	  prefix: List<&2, SnapshotResource>,
1119	  tail: List<&2, SnapshotResource>,
1120	  reservations: Registry
1121	) -> OwnerSnapshotCloseDecision:
1122	  match copies:
1123	    case (for_check, for_update):
1124	      match for_check:
1125	        case SnapshotResource{_, _, retained_bytes, _}:
1126	          owner.snapshot.close.released(
1127	            registry.release(reservations, retained_bytes), allocator,
1128	            cursors, prefix, for_update, tail)
1129	
1130	def owner.snapshot.close.list(
1131	  allocator: ResourceIdAllocator,
1132	  reservations: Registry,
1133	  cursors: List<&2, CursorResource>,
1134	  decision: SnapshotListCursorDecision
1135	) -> OwnerSnapshotCloseDecision:
1136	  match decision:
1137	    case SnapshotListClosed{prefix, resource, tail}:
1138	      owner.snapshot.close.resource(snapshot.resource.duplicate(resource),
1139	        allocator, cursors, prefix, tail, reservations)
1140	    case SnapshotListBusy{snapshots}:
1141	      OwnerSnapshotBusy{OwnerRegistry{allocator, reservations, snapshots,
1142	        cursors}}
1143	    case SnapshotListRejected{snapshots}:
1144	      OwnerSnapshotCloseRejected{OwnerRegistry{allocator, reservations,
1145	        snapshots, cursors}}
1146	    case SnapshotListUpdated{snapshots}:
1147	      OwnerSnapshotCloseRejected{OwnerRegistry{allocator, reservations,
1148	        snapshots, cursors}}
1149	
1150	# Closes a snapshot only after its final cursor releases the retained root.
1151	def owner.snapshot.close(
1152	  owner: OwnerRegistry,
1153	  snapshot_id: U32
1154	) -> OwnerSnapshotCloseDecision:
1155	  match owner:
1156	    case OwnerRegistry{allocator, reservations, snapshots, cursors}:
1157	      owner.snapshot.close.list(allocator, reservations, cursors,
1158	        snapshot.list.cursor.close(snapshots, snapshot_id))
1159	
1160	# Maps a valid registry release to successful snapshot closure.
1161	def snapshot.close.registry(decision: RegistryReleaseDecision) -> SnapshotCloseDecision:
1162	  match decision:
1163	    case Released{registry}: SnapshotClosed{registry}
1164	    case InvalidRelease{registry}: InvalidSnapshotClose{registry}
1165	
1166	# Leaves registry accounting untouched while a cursor still references the snapshot.
1167	def snapshot.close(
1168	  lifecycle: Lifecycle,
1169	  registry: Registry,
1170	  released_bytes: Nat
1171	) -> SnapshotCloseDecision:
1172	  match lifecycle:
1173	    case Closed{}: InvalidSnapshotClose{registry}
1174	    case Live{cursor_count}:
1175	      match cursor_count:
1176	        case 0n:
1177	          snapshot.close.registry(registry.release(registry, released_bytes))
1178	        case 1n+rest: SnapshotBusy{registry}
1179	
1180	# Adds a cursor reference to an open snapshot.
1181	def cursor.acquire(lifecycle: Lifecycle) -> LifecycleDecision:
1182	  match lifecycle:
1183	    case Live{cursor_count}: Advanced{Live{Nat.add(cursor_count, 1n)}}
1184	    case Closed{}: Invalid{}
1185	
1186	# Releases one cursor reference, rejecting underflow and closed snapshots.
1187	def cursor.release(lifecycle: Lifecycle) -> LifecycleDecision:
1188	  match lifecycle:
1189	    case Live{cursor_count}:
1190	      match cursor_count:
1191	        case 0n: Invalid{}
1192	        case 1n+rest: Advanced{Live{rest}}
1193	    case Closed{}: Invalid{}
1194	
1195	# Closes a snapshot only after its final cursor reference has been released.
1196	def close(lifecycle: Lifecycle) -> LifecycleDecision:
1197	  match lifecycle:
1198	    case Live{cursor_count}:
1199	      match cursor_count:
1200	        case 0n: Advanced{Closed{}}
1201	        case 1n+rest: Busy{}
1202	    case Closed{}: Invalid{}
1203	
1204	# Publishes a generation from independent copies of its current identifier.
1205	def publish_generation.copied(
1206	  copies: Nat & Nat,
1207	  current_bytes: Nat,
1208	  references: Nat
1209	) -> GenerationDecision:
1210	  match copies:
1211	    case (next_source, retained_source):
1212	      match references:
1213	        case 0n: Published{Nat.add(next_source, 1n), NoRetained{}}
1214	        case 1n+rest:
1215	          Published{Nat.add(next_source, 1n),
1216	            GenerationRetained{retained_source, current_bytes, 1n+rest}}
1217	
1218	# Publishes a new current generation and retains its predecessor while referenced.
1219	def publish_generation(
1220	  current: Nat,
1221	  current_bytes: Nat,
1222	  references: Nat
1223	) -> GenerationDecision:
1224	  publish_generation.copied(nat.duplicate(current), current_bytes, references)
1225	
1226	def publish.registry.current(
1227	  current: FileGeneration,
1228	  retired: List<&2, FileGeneration>,
1229	  next_bytes: Nat
1230	) -> GenerationRegistry:
1231	  match current:
1232	    case FileGeneration{+id, +size, references}:
1233	      match references:
1234	        case 0n:
1235	          GenerationRegistry{FileGeneration{Nat.add(id, 1n),
1236	            next_bytes, 0n}, retired}
1237	        case 1n+rest:
1238	          GenerationRegistry{FileGeneration{Nat.add(id, 1n),
1239	            next_bytes, 0n}, Con{FileGeneration{id, size, 1n+rest}, retired}}
1240	
1241	def publish.registry(
1242	  registry: GenerationRegistry,
1243	  next_bytes: Nat
1244	) -> GenerationRegistry:
1245	  match registry:
1246	    case GenerationRegistry{current, retired}:
1247	      publish.registry.current(current, retired, next_bytes)
1248	
1249	def pin.retired.prepend(
1250	  id: Nat,
1251	  size: Nat,
1252	  references: Nat,
1253	  result: RetiredPin
1254	) -> RetiredPin:
1255	  match result:
1256	    case RetiredPinned{retired}:
1257	      RetiredPinned{Con{FileGeneration{id, size, references}, retired}}
1258	    case RetiredPinMissing{retired}:
1259	      RetiredPinMissing{Con{FileGeneration{id, size, references}, retired}}
1260	
1261	def pin.retired.loop(
1262	  retired: List<&2, FileGeneration>,
1263	  +target: Nat,
1264	  matches: Bool
1265	) -> RetiredPin:
1266	  match retired:
1267	    case Nil{}: RetiredPinMissing{Nil{}}
1268	    case Con{FileGeneration{+id, +size, +references}, Nil{}}:
1269	      match matches:
1270	        case True{}:
1271	          RetiredPinned{Con{FileGeneration{id, size,
1272	            Nat.add(references, 1n)}, Nil{}}}
1273	        case False{}:
1274	          RetiredPinMissing{Con{FileGeneration{id, size,
1275	            references}, Nil{}}}
1276	    case Con{FileGeneration{+id, +size, +references},
1277	        Con{FileGeneration{+next_id, +next_size, +next_references}, rest}}:
1278	      match matches:
1279	        case True{}:
1280	          RetiredPinned{Con{FileGeneration{id, size,
1281	            Nat.add(references, 1n)},
1282	            Con{FileGeneration{next_id, next_size, next_references}, rest}}}
1283	        case False{}:
1284	          pin.retired.prepend(id, size, references,
1285	            pin.retired.loop(Con{FileGeneration{next_id, next_size,
1286	              next_references}, rest}, target,
1287	              Nat.is_eq(next_id, target)))
1288	
1289	def pin.retired(
1290	  +retired: List<&2, FileGeneration>,
1291	  +target: Nat
1292	) -> RetiredPin:
1293	  match retired:
1294	    case Nil{}: RetiredPinMissing{Nil{}}
1295	    case Con{generation, _}:
1296	      match generation:
1297	        case FileGeneration{+id, _, _}:
1298	          pin.retired.loop(retired, target, Nat.is_eq(id, target))
1299	
1300	def pin.registry.retired(
1301	  id: Nat,
1302	  size: Nat,
1303	  references: Nat,
1304	  retired: RetiredPin
1305	) -> GenerationPinDecision:
1306	  match retired:
1307	    case RetiredPinned{retired}:
1308	      Pinned{GenerationRegistry{FileGeneration{id, size, references}, retired}}
1309	    case RetiredPinMissing{retired}:
1310	      MissingGeneration{GenerationRegistry{FileGeneration{id, size,
1311	        references}, retired}}
1312	
1313	def pin.registry.decision(
1314	  +id: Nat,
1315	  +size: Nat,
1316	  +references: Nat,
1317	  +retired: List<&2, FileGeneration>,
1318	  +target: Nat,
1319	  matches: Bool
1320	) -> GenerationPinDecision:
1321	  match matches:
1322	    case True{}:
1323	      Pinned{GenerationRegistry{FileGeneration{id, size,
1324	        Nat.add(references, 1n)}, retired}}
1325	    case False{}:
1326	      pin.registry.retired(id, size, references,
1327	        pin.retired(retired, target))
1328	
1329	def pin.registry.current(
1330	  current: FileGeneration,
1331	  retired: List<&2, FileGeneration>,
1332	  +target: Nat
1333	) -> GenerationPinDecision:
1334	  match current:
1335	    case FileGeneration{+id, +size, references}:
1336	      pin.registry.decision(id, size, references, retired, target,
1337	        Nat.is_eq(id, target))
1338	
1339	# Increments the reference count of one current or retired generation.
1340	def pin_generation(
1341	  registry: GenerationRegistry,
1342	  target: Nat
1343	) -> GenerationPinDecision:
1344	  match registry:
1345	    case GenerationRegistry{current, retired}:
1346	      pin.registry.current(current, retired, target)
1347	
1348	def release.retired.prepend(
1349	  id: Nat,
1350	  size: Nat,
1351	  references: Nat,
1352	  result: RetiredRelease
1353	) -> RetiredRelease:
1354	  match result:
1355	    case RetiredReleased{retired}:
1356	      RetiredReleased{Con{FileGeneration{id, size, references}, retired}}
1357	    case RetiredReclaimed{retired, number, bytes}:
1358	      RetiredReclaimed{Con{FileGeneration{id, size, references}, retired},
1359	        number, bytes}
1360	    case RetiredMissing{retired}:
1361	      RetiredMissing{Con{FileGeneration{id, size, references}, retired}}
1362	
1363	def release.retired.references(
1364	  id: Nat,
1365	  size: Nat,
1366	  references: Nat,
1367	  tail: List<&2, FileGeneration>
1368	) -> RetiredRelease:
1369	  match references:
1370	    case 0n:
1371	      RetiredMissing{Con{FileGeneration{id, size, references}, tail}}
1372	    case 1n:
1373	      RetiredReclaimed{tail, id, size}
1374	    case 1n+rest:
1375	      RetiredReleased{Con{FileGeneration{id, size, rest}, tail}}
1376	
1377	def release.retired.loop(
1378	  retired: List<&2, FileGeneration>,
1379	  +target: Nat,
1380	  matches: Bool
1381	) -> RetiredRelease:
1382	  match retired:
1383	    case Nil{}: RetiredMissing{Nil{}}
1384	    case Con{FileGeneration{+id, +size, +references}, Nil{}}:
1385	      match matches:
1386	        case True{}:
1387	          release.retired.references(id, size, references, Nil{})
1388	        case False{}:
1389	          RetiredMissing{Con{FileGeneration{id, size,
1390	            references}, Nil{}}}
1391	    case Con{FileGeneration{+id, +size, +references},
1392	        Con{FileGeneration{+next_id, +next_size, +next_references}, rest}}:
1393	      match matches:
1394	        case True{}:
1395	          release.retired.references(id, size, references,
1396	            Con{FileGeneration{next_id, next_size, next_references}, rest})
1397	        case False{}:
1398	          release.retired.prepend(id, size, references,
1399	            release.retired.loop(Con{FileGeneration{next_id, next_size,
1400	              next_references}, rest}, target,
1401	              Nat.is_eq(next_id, target)))
1402	
1403	def release.retired(
1404	  +retired: List<&2, FileGeneration>,
1405	  +target: Nat
1406	) -> RetiredRelease:
1407	  match retired:
1408	    case Nil{}: RetiredMissing{Nil{}}
1409	    case Con{generation, _}:
1410	      match generation:
1411	        case FileGeneration{+id, _, _}:
1412	          release.retired.loop(retired, target, Nat.is_eq(id, target))
1413	
1414	def release.registry.retired(
1415	  id: Nat,
1416	  size: Nat,
1417	  references: Nat,
1418	  result: RetiredRelease
1419	) -> GenerationReleaseDecision:
1420	  match result:
1421	    case RetiredReleased{retired}:
1422	      GenerationReleased{GenerationRegistry{
1423	        FileGeneration{id, size, references}, retired}}
1424	    case RetiredReclaimed{retired, number, bytes}:
1425	      GenerationReclaimed{GenerationRegistry{
1426	        FileGeneration{id, size, references}, retired}, number, bytes}
1427	    case RetiredMissing{retired}:
1428	      MissingReference{GenerationRegistry{
1429	        FileGeneration{id, size, references}, retired}}
1430	
1431	def release.registry.references(
1432	  id: Nat,
1433	  size: Nat,
1434	  references: Nat,
1435	  retired: List<&2, FileGeneration>
1436	) -> GenerationReleaseDecision:
1437	  match references:
1438	    case 0n:
1439	      MissingReference{GenerationRegistry{
1440	        FileGeneration{id, size, references}, retired}}
1441	    case 1n+rest:
1442	      GenerationReleased{GenerationRegistry{FileGeneration{id, size, rest},
1443	        retired}}
1444	
1445	def release.registry.decision(
1446	  +id: Nat,
1447	  +size: Nat,
1448	  +references: Nat,
1449	  +retired: List<&2, FileGeneration>,
1450	  +target: Nat,
1451	  matches: Bool
1452	) -> GenerationReleaseDecision:
1453	  match matches:
1454	    case True{}:
1455	      release.registry.references(id, size, references, retired)
1456	    case False{}:
1457	      release.registry.retired(id, size, references,
1458	        release.retired(retired, target))
1459	
1460	def release.registry.current(
1461	  current: FileGeneration,
1462	  retired: List<&2, FileGeneration>,
1463	  +target: Nat
1464	) -> GenerationReleaseDecision:
1465	  match current:
1466	    case FileGeneration{+id, +size, references}:
1467	      release.registry.decision(id, size, references, retired, target,
1468	        Nat.is_eq(id, target))
1469	
1470	def release_generation.in(
1471	  registry: GenerationRegistry,
1472	  target: Nat
1473	) -> GenerationReleaseDecision:
1474	  match registry:
1475	    case GenerationRegistry{current, retired}:
1476	      release.registry.current(current, retired, target)
1477	
1478	def release_generation.from_registry(
1479	  registry: GenerationRegistry,
1480	  target: Nat
1481	) -> GenerationReleaseDecision:
1482	  release_generation.in(registry, target)
1483	
1484	# Releases one reference and reclaims a retired generation after its final reader.
1485	def release_generation(retired: RetiredGeneration) -> RetentionDecision:
1486	  match retired:
1487	    case NoRetained{}: InvalidReference{}
1488	    case GenerationRetained{generation, bytes, references}:
1489	      match references:
1490	        case 0n: InvalidReference{}
1491	        case 1n: Reclaim{generation, bytes}
1492	        case 1n+rest: Keep{generation, bytes, rest}