src/scan/Snapshot.bend

mylsm-lsm-store@0.6.0.1 · 1,492 lines · 137 definitions

1import Base
2import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
3
4# Reports whether the requested snapshot fits the retained-resource limits.
5type Decision is Data:
6  Retained{snapshot_count: Nat, retained_bytes: Nat}
7  ResourceLimit{}
8
9# Holds the aggregate reservation counters for one database owner.
10type Registry is Data:
11  Registry{snapshot_count: Nat, retained_bytes: Nat}
12
13# Returns the updated registry or its unchanged prior state after rejection.
14type RegistryDecision is Data:
15  Admitted{registry: Registry}
16  Rejected{registry: Registry}
17
18# Reports a snapshot release or preserves the registry on invalid release.
19type RegistryReleaseDecision is Data:
20  Released{registry: Registry}
21  InvalidRelease{registry: Registry}
22
23# Couples snapshot closure to child checks and aggregate reservation release.
24type 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.
30type Lifecycle is Data:
31  Live{cursor_count: Nat}
32  Closed{}
33
34# Reports a lifecycle transition, a live-child conflict, or an invalid action.
35type LifecycleDecision is Data:
36  Advanced{lifecycle: Lifecycle}
37  Busy{}
38  Invalid{}
39
40# Reports a published generation and whether its predecessor remains retained.
41type GenerationDecision is Data:
42  Published{current: Nat, retired: RetiredGeneration}
43
44# Represents one older SST generation while snapshots still reference it.
45type 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.
50type 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.
56type FileGeneration is Data:
57  FileGeneration{id: Nat, size: Nat, references: Nat}
58
59# Keeps the current generation separate from all reader-retained generations.
60type 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.
65type GenerationPinDecision is Data:
66  Pinned{registry: GenerationRegistry}
67  MissingGeneration{registry: GenerationRegistry}
68
69# Reports an incremented reader count for a located retired generation.
70type 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.
75type 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.
81type 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.
87type ResourceIdAllocator is Data:
88  ResourceIdAllocator{next_id: U32, exhausted: Bool}
89
90# Reports an allocated owner identifier or preserves an exhausted allocator.
91type 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.
96type 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.
101type CursorResource is Data:
102  CursorResource{id: U32, snapshot_id: U32}
103
104# Holds aggregate snapshot reservations and their owner-issued identifiers.
105type 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.
110type 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.
115type 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.
120type 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.
127type 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.
138type 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.
143type 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.
149type 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.
157type 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.
163type 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.
176type SnapshotCursorDecision is Data:
177  SnapshotCursorAcquired{resource: SnapshotResource}
178  SnapshotCursorLimit{resource: SnapshotResource}
179  SnapshotCursorReleased{resource: SnapshotResource}
180  SnapshotCursorUnderflow{resource: SnapshotResource}
181
182def point_read.matches(order: Cmp) -> Bool:
183  match order:
184    case EQ{}: True{}
185    case _: False{}
186
187
188
189def 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
195def nat.duplicate(value: Nat) -> Nat & Nat:
196  nat.duplicate.go(value, 0n, 0n)
197
198def 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
212def u32.bytes(value: U32) -> Bytes.Bytes:
213  Bytes.set.u32be(Bytes.new(4), 0, value)
214
215def 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
222def 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
228def 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
236def 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
242def u32.duplicate(value: U32) -> U32 & U32:
243  u32.duplicate.encoded(u32.bytes(value))
244
245def 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.
259def 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.
267def 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.
279def 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.
298def 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.
304def 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.
316def 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.
323def 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.
336def 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.
350def 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.
362def 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.
375def 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.
396def 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.
407def 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.
418def 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
424def 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
438def 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
454def 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
470def 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
488def 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
507def 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
529def 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
541def 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
559def 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
566def 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
578def 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
586def 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.
598def 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
605def 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
617def 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
624def 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.
634def snapshot.resource.cursor.release(
635  resource: SnapshotResource
636) -> SnapshotCursorDecision:
637  snapshot.resource.cursor.release.copied(
638    snapshot.resource.duplicate(resource))
639
640def 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
649def 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
661def 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
666def 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
685def 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
697def 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
709def 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
726def 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.
742def 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
783def 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.
795def 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.
805def 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.
814def 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
822def 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
835def 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
850def 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
871def 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.
887def 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.
905def 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
912def 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
923def 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
929def 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
938def 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
949def cursor.list.length.keep(
950  cursors: List<&2, CursorResource>
951) -> List<&2, CursorResource> & Nat:
952  cursor.list.length.keep.go(cursors, Nil{}, 0n)
953
954def 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
968def 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
1002def 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.
1012def 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
1018def 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
1044def 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
1059def 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
1074def 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.
1089def 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
1098def 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
1114def 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
1130def 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.
1151def 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.
1161def 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.
1167def 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.
1181def 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.
1187def 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.
1196def 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.
1205def 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.
1219def 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
1226def 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
1241def 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
1249def 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
1261def 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
1289def 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
1300def 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
1313def 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
1329def 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.
1340def 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
1348def 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
1363def 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
1377def 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
1403def 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
1414def 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
1431def 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
1445def 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
1460def 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
1470def 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
1478def 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.
1485def 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}