1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
use super::{Dot, DotSet, DotSetStation, StationDots};
impl DotSet {
/// Number of nonempty station rows.
#[must_use]
pub fn station_count(&self) -> usize {
self.stations.len()
}
/// Allocation-free summaries of nonempty station rows in station order.
pub fn stations(&self) -> impl Iterator<Item = DotSetStation> + '_ {
self.stations.iter().map(|(&station, slot)| DotSetStation {
station,
floor: slot.floor,
high_water: slot.high_water(),
})
}
/// Allocation-free summary of one station row.
#[must_use]
pub fn station(&self, station: u32) -> Option<DotSetStation> {
self.stations.get(&station).map(|slot| DotSetStation {
station,
floor: slot.floor,
high_water: slot.high_water(),
})
}
/// Crate-internal (S205): the set's occupancy pages, per
/// `(station, page-of-64)`, each with its `u64` bit mask (bit `k` is
/// dot `page * 64 + k`), in ascending page order with no zero masks.
///
/// Computed off the compact form directly, `O(pages + exceptions)`
/// and never a per-dot walk of the prefix: a page fully inside the
/// floor is one all-ones word (page zero minus the non-dot bit), the
/// floor's partial page is one shift, and the parked exceptions fold
/// into their pages as bits. Adjacent rows may share a key where the
/// floor's partial page meets its first exceptions (and between
/// exceptions sharing a page); the consumer coalesces them, so the
/// walk stays lazy and allocation-free. The rhapsody's paged
/// possession shell (`rhapsody/possession/mod.rs`) builds from this walk;
/// it is not a public read because the page width is that shell's
/// factoring, not this type's contract.
pub(crate) fn occupancy_pages(&self) -> impl Iterator<Item = ((u32, u64), u64)> + '_ {
self.stations.iter().flat_map(|(&station, slot)| {
let full_floor_pages = slot.floor / 64;
// The floor's own pages: full words below, one shifted word
// at the partial page (absent when the floor sits exactly on
// a page boundary), bit zero of page zero permanently clear
// (dot 0 is the non-dot).
let floor_pages = (0..=full_floor_pages)
.map(move |page| {
let mask = if page < full_floor_pages {
!0u64
} else {
let high = slot.floor % 64;
if slot.floor == 0 {
0
} else if high == 63 {
!0u64
} else {
(1u64 << (high + 1)) - 1
}
};
(page, if page == 0 { mask & !1 } else { mask })
})
.filter(|&(_, mask)| mask != 0);
// The parked exceptions, grouped into their pages: all lie
// strictly above `floor + 1`, so their pages are at or past
// the floor's partial page; a shared boundary page merges in
// the coalescing scan below.
let exception_pages = slot.above.iter().map(|&dot| (dot / 64, 1u64 << (dot % 64)));
// Both streams ascend within the station; rows may repeat a
// page at the floor/exception boundary and between exceptions
// sharing a page, and the consumer coalesces equal-adjacent
// keys (the walk itself stays lazy and allocation-free, the
// house posture for these reads).
floor_pages
.chain(exception_pages)
.map(move |(page, mask)| ((station, page), mask))
})
}
/// The *holes*: every missing dot below the high water, in ascending
/// [`Dot`] order.
///
/// Exactly the difference between the have-set's down-closure and the
/// have-set: the fetch list a repair protocol serves to raise the floor.
/// Lazy and allocation-free; the hole count is bounded by the actual gap
/// spans, which an adversarially high dot can make enormous, so a consumer
/// materializing this read should bound its own take.
pub fn holes(&self) -> impl Iterator<Item = Dot> + '_ {
self.stations.iter().flat_map(|(&station, slot)| {
slot.above
.iter()
.scan(slot.floor.saturating_add(1), |expected, &dot| {
let gap = *expected..dot;
*expected = dot.saturating_add(1);
Some(gap)
})
.flatten()
// A hole lies strictly above the floor, so it starts at
// `floor + 1 >= 1`: the counter is nonzero by construction
// and the filter drops nothing (ruling R-91).
.filter_map(move |counter| Dot::from_parts(station, counter).ok())
})
}
/// The *owed dots*: every dot in `self` that `other` lacks, in ascending
/// [`Dot`] order.
///
/// The difference read at dot granularity (PRD 0014). Where
/// [`VersionVector::difference`](crate::metis::VersionVector::difference)
/// names the *claims* a creditor must be raised to, this enumerates the
/// exact transfer list, the anti-entropy answer to "what do I owe you":
/// serving these dots to a replica holding `other` converges it to the two
/// sides' [`merge`](Self::merge), and between genuine cuts (both sides
/// [`from_witnessed`](Self::from_witnessed)) it yields exactly the spans
/// `(other.get(s), self.get(s)]` the vector residual names.
///
/// The two shipped asymmetries are its special cases, pinned by property
/// test: [`holes`](Self::holes) is the difference of the down-closure
/// against the set (`from_cut(&self.high_water()).difference(&self)`, the
/// fetch list), and the parked exceptions are the difference of the set
/// against its interior (`self.difference(&from_cut(&self.floor()))`, the
/// reorder exposure). What to fetch and what to serve are one read,
/// pointed in opposite directions.
///
/// Lazy and allocation-free like [`holes`](Self::holes), and unlike the
/// holes it cannot be inflated by an adversarially high dot: it yields
/// only dots `self` actually holds. Batching, capping, and scheduling a
/// transfer built from it are the caller's policy. Commutes exactly with
/// [`restrict`](Self::restrict) (the PRD 0013 R2 base-change law): the
/// owed dots of the restrictions are the owed dots at the roster's
/// stations.
pub fn difference<'a>(&'a self, other: &'a Self) -> impl Iterator<Item = Dot> + 'a {
self.stations.iter().flat_map(move |(&station, slot)| {
let peer = other.stations.get(&station);
let peer_floor = peer.map_or(0, |dots| dots.floor);
// `peer` is a `Copy` capture, so the closure is `Copy` and serves
// both halves of the chain.
let owed = move |dot: &u64| !peer.is_some_and(|dots| dots.contains(*dot));
let prefix = (peer_floor < slot.floor)
.then(|| peer_floor.saturating_add(1)..=slot.floor)
.into_iter()
.flatten()
.filter(owed);
let parked = slot.above.iter().copied().filter(owed);
// Both legs yield held counters, which are one-based: the prefix
// starts at `peer_floor + 1 >= 1` and an exception is strictly
// above its floor, so the filter drops nothing (ruling R-91).
prefix
.chain(parked)
.filter_map(move |counter| Dot::from_parts(station, counter).ok())
})
}
/// Whether this have-set contains every dot `other` holds: whole-set
/// coverage, true iff `other.difference(self)` is empty.
///
/// The coverage predicate the kernel and delta reads need without paying
/// [`difference`](Self::difference)'s per-dot enumeration: it decides each
/// station on the floors and exception runs alone (a range count for the
/// prefix tail, a membership probe per exception), so a deep gap-free
/// prefix costs `O(stations + exceptions)` rather than its cardinality.
///
/// The law it stands on, pinned by property test:
/// `a.covers(b) == b.difference(a).next().is_none()`. A station `other`
/// holds that `self` lacks entirely fails coverage (canonical form keeps no
/// empty station, so a present station has a dot to miss); conversely every
/// station `self` holds beyond `other` is irrelevant, coverage being
/// one-directional.
#[must_use]
pub fn covers(&self, other: &Self) -> bool {
other.stations.iter().all(|(station, their_slot)| {
self.stations
.get(station)
.is_some_and(|my_slot| my_slot.covers(their_slot))
})
}
/// The number of dots parked out of order (above some unfilled hole),
/// across all stations.
///
/// The observability read for the type's one unbounded axis: in-order
/// receipt holds it at `0`, and a caller bounding reorder exposure watches
/// it the way it watches `pending_len`. Memory is `O(stations +
/// exceptions)`.
#[must_use]
pub fn exceptions_len(&self) -> usize {
self.stations
.values()
.fold(0, |count, slot| count.saturating_add(slot.above.len()))
}
/// Iterates the out-of-order dots above each station's gap-free floor in
/// ascending [`Dot`] order.
///
/// This is the sparse half of the compact representation. Consumers with
/// an ordered dot index combine it with [`floor`](Self::floor): range-query
/// each floor, then probe these exceptions exactly, avoiding a scan up to a
/// potentially adversarial high water. The walk is allocation-free and
/// yields exactly [`exceptions_len`](Self::exceptions_len) dots.
pub fn exceptions(&self) -> impl Iterator<Item = Dot> + '_ {
self.stations.iter().flat_map(|(&station, slot)| {
// An exception is parked strictly above its station's floor, so
// its counter is nonzero and the filter drops nothing (R-91).
slot.above
.iter()
.filter_map(move |&counter| Dot::from_parts(station, counter).ok())
})
}
/// The greatest received dot for one station (`0` when none has been).
///
/// The per-station coordinate of [`high_water`](Self::high_water), without
/// materializing the whole vector: the read a dot-minting site consults
/// (`Dotted::next_dot` is `high_water_of + 1`, off the *context*, which
/// never forgets; minting off a store that forgets removals would re-issue
/// a superseded dot).
#[must_use]
pub fn high_water_of(&self, station: u32) -> u64 {
self.stations
.get(&station)
.map_or(0, StationDots::high_water)
}
/// The *holes for one station*: every missing dot below that station's high
/// water, ascending.
///
/// The per-station view of [`holes`](Self::holes), the repair loop's read
/// (PRD 0012's first open question, additive under R7). Equal to the
/// whole-set [`holes`](Self::holes) filtered to `station`: same order, same
/// dots, and empty for a station with no received dots or no gap below its
/// high water. Lazy and allocation-free like the whole-set read, and like it
/// inflatable by an adversarially high dot, so a materializing caller should
/// bound its own take.
pub fn holes_for(&self, station: u32) -> impl Iterator<Item = u64> + '_ {
self.stations.get(&station).into_iter().flat_map(|slot| {
slot.above
.iter()
.scan(slot.floor.saturating_add(1), |expected, &dot| {
let gap = *expected..dot;
*expected = dot.saturating_add(1);
Some(gap)
})
.flatten()
})
}
/// The *hole count*: the total number of missing dots below the high water
/// across all stations.
///
/// The repair-debt norm (PRD 0012's third open question, additive under R7),
/// equal to `holes().count()` but computed in `O(stations + exceptions)`
/// arithmetic without enumerating a single gap: per station the high water
/// minus the floor is the span above the prefix, and the parked exceptions
/// are the dots in that span already held, so the holes are their difference.
/// Saturating at the [`u64`] ceiling like the rest of the crate; an
/// adversarially high dot makes the true count enormous, and this reports
/// [`u64::MAX`] rather than overflowing.
#[must_use]
pub fn hole_count(&self) -> u64 {
self.stations.values().fold(0u64, |total, slot| {
// The span strictly above the floor and up to the high water; every
// dot in it is either a hole or a parked exception, and the two
// partition the span, so holes = span - exceptions.
let span = slot.high_water().saturating_sub(slot.floor);
let held = u64::try_from(slot.above.len()).unwrap_or(u64::MAX);
total.saturating_add(span.saturating_sub(held))
})
}
/// Every held dot, in ascending [`Dot`] order: each station's gap-free
/// prefix, then its parked exceptions.
///
/// The enumeration read: what the causal store axis consumes
/// ([`DotStore::dots`](crate::metis::DotStore::dots) on the bare store
/// delegates here) and what a consumer reconciling two contexts by hand
/// reaches for. Lazy and allocation-free; the yield count is the set's
/// cardinality (the floor prefix is enumerated dot by dot), which is
/// intrinsic to the answer, so a caller walking a deep floor should
/// prefer the cut reads ([`floor`](Self::floor),
/// [`high_water`](Self::high_water)) where a summary suffices.
pub fn dots(&self) -> impl Iterator<Item = Dot> + '_ {
self.stations.iter().flat_map(|(&station, slot)| {
// The prefix starts at dot one and every exception is strictly
// above its floor, so each counter is nonzero and the filter
// drops nothing (ruling R-91).
(1..=slot.floor)
.chain(slot.above.iter().copied())
.filter_map(move |counter| Dot::from_parts(station, counter).ok())
})
}
}