• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

prisma-risk / tsoracle / 26491073736

27 May 2026 04:37AM UTC coverage: 94.936% (-0.01%) from 94.949%
26491073736

push

github

web-flow
fix(core/allocator): try_prepare_window_extension +1 overflow guard reported wrong operand (#565)

The `committed_high_water.checked_add(1)?` guard reported
`PhysicalMsOutOfRange(*committed_high_water)` on overflow — but that
operand is in range; the value that actually went out of range is
`committed_high_water + 1`. A reader of the error would see an in-range
number and infer the wrong root cause.

The branch is also dead: `try_on_leadership_gained` rejects
`committed_ceiling > PHYSICAL_MS_MAX` and `try_commit_window_extension`
does the same for any subsequent advance, so `committed_high_water` is
bounded at `PHYSICAL_MS_MAX = (1 << 46) - 1`, many orders of magnitude
below `u64::MAX`. The +1 cannot overflow without an upstream invariant
violation.

Replace with a `debug_assert!` that documents the invariant and trips
loudly in debug if it is ever broken. Release-mode safety is preserved
by the existing downstream `requested > PHYSICAL_MS_MAX` check at
line 463: a hypothetical wraparound would surface there.

Signed-off-by: Sebastian Thiebaud <sebastian@prismarisk.com>

3 of 5 new or added lines in 1 file covered. (60.0%)

13760 of 14494 relevant lines covered (94.94%)

379505.54 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

99.16
/crates/tsoracle-core/src/allocator.rs
1
//
2
//  ░▀█▀░█▀▀░█▀█░█▀▄░█▀█░█▀▀░█░░░█▀▀
3
//  ░░█░░▀▀█░█░█░█▀▄░█▀█░█░░░█░░░█▀▀
4
//  ░░▀░░▀▀▀░▀▀▀░▀░▀░▀░▀░▀▀▀░▀▀▀░▀▀▀
5
//
6
//  tsoracle — Distributed Timestamp Oracle
7
//  https://www.tsoracle.rs
8
//
9
//  Copyright (c) 2026 Prisma Risk
10
//
11
//  Licensed under the Apache License, Version 2.0 (the "License");
12
//  you may not use this file except in compliance with the License.
13
//  You may obtain a copy of the License at
14
//
15
//      https://www.apache.org/licenses/LICENSE-2.0
16
//
17
//  Unless required by applicable law or agreed to in writing, software
18
//  distributed under the License is distributed on an "AS IS" BASIS,
19
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
20
//  See the License for the specific language governing permissions and
21
//  limitations under the License.
22
//
23

24
// #[PerformanceCriticalPath]
25
//! The window allocator state machine. Sync, no I/O.
26

27
use crate::{Epoch, LOGICAL_MAX, PHYSICAL_MS_MAX, Timestamp};
28

29
/// A `u64` physical-millisecond value proven `<= PHYSICAL_MS_MAX` at
30
/// construction.
31
///
32
/// The 46-bit physical field of [`Timestamp`] cannot represent any value above
33
/// [`PHYSICAL_MS_MAX`]; every allocator entry point used to re-check that
34
/// bound on bare `u64` parameters (`fence_floor`, `committed_ceiling`,
35
/// `now_ms`, `persisted_high_water`) — three different methods, each carrying
36
/// its own `if value > PHYSICAL_MS_MAX { ... }` line. `PhysicalMs` collapses
37
/// those per-method runtime checks into one construction-site check, so:
38
///
39
/// * a method signature taking `PhysicalMs` is compile-time proof that the
40
///   46-bit bound has already been validated for that argument; and
41
/// * an accidental swap of `now_ms` and `committed_ceiling` at a call site no
42
///   longer type-checks against bare `u64` clocks, durations, or counters.
43
///
44
/// Constructed via [`try_new`](Self::try_new) (or the equivalent
45
/// [`TryFrom<u64>`] impl). The inner value can be recovered with
46
/// [`get`](Self::get) for arithmetic; the result must be re-wrapped through
47
/// `try_new` before crossing back into a `PhysicalMs`-typed boundary.
48
#[derive(Copy, Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
49
pub struct PhysicalMs(u64);
50

51
impl PhysicalMs {
52
    /// The largest in-range value, `2^46 - 1` (equal to [`PHYSICAL_MS_MAX`]).
53
    pub const MAX: PhysicalMs = PhysicalMs(PHYSICAL_MS_MAX);
54
    /// The zero value. Available as `const`, matching `Duration::ZERO` style.
55
    pub const ZERO: PhysicalMs = PhysicalMs(0);
56

57
    /// Validate `value <= PHYSICAL_MS_MAX` and wrap. Returns
58
    /// [`CoreError::PhysicalMsOutOfRange`] otherwise.
59
    ///
60
    /// Declared `const fn` so [`MAX`](Self::MAX) and any other compile-time
61
    /// `PhysicalMs` constant can be built without unsafe direct-field
62
    /// construction outside this module.
63
    pub const fn try_new(value: u64) -> Result<Self, CoreError> {
25,249✔
64
        if value > PHYSICAL_MS_MAX {
25,249✔
65
            return Err(CoreError::PhysicalMsOutOfRange(value));
5✔
66
        }
25,244✔
67
        Ok(PhysicalMs(value))
25,244✔
68
    }
25,249✔
69

70
    /// Recover the underlying `u64`. `Copy`, so the receiver remains usable.
71
    pub const fn get(self) -> u64 {
25,244✔
72
        self.0
25,244✔
73
    }
25,244✔
74
}
75

76
impl TryFrom<u64> for PhysicalMs {
77
    type Error = CoreError;
78
    fn try_from(value: u64) -> Result<Self, Self::Error> {
2✔
79
        Self::try_new(value)
2✔
80
    }
2✔
81
}
82

83
impl core::fmt::Display for PhysicalMs {
84
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
1✔
85
        core::fmt::Display::fmt(&self.0, f)
1✔
86
    }
1✔
87
}
88

89
/// A contiguous block of `count` timestamps starting at
90
/// `(physical_ms, logical_start)`, all sharing one leadership `epoch`.
91
///
92
/// Fields are private and the only public constructor is
93
/// [`try_new`](Self::try_new), which validates that every timestamp the grant
94
/// covers fits the packed 46-bit physical / 18-bit logical layout. A value of
95
/// this type is therefore proof that [`first`](Self::first) and
96
/// [`last`](Self::last) can pack without panicking — the in-range invariant is
97
/// guaranteed by the type, not by the one constructor that happens to build it.
98
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
99
pub struct WindowGrant {
100
    physical_ms: u64,
101
    logical_start: u32,
102
    count: u32,
103
    epoch: Epoch,
104
}
105

106
impl WindowGrant {
107
    /// Construct a grant, checking that every timestamp it covers packs
108
    /// cleanly. This is the only way to build a `WindowGrant` outside this
109
    /// module, so a constructed value witnesses that `first`/`last` are
110
    /// infallible.
111
    ///
112
    /// Rejects `count == 0` (a grant covers at least one timestamp, and
113
    /// `last`'s `logical_start + count - 1` would underflow). The range check
114
    /// defers to [`Timestamp::try_pack`] on the *last* logical the grant emits:
115
    /// it is the single source of truth for the bit layout, and since
116
    /// `logical_start <= last_logical <= LOGICAL_MAX` validating the last
117
    /// boundary validates the first by implication.
118
    pub fn try_new(
13,787✔
119
        physical_ms: u64,
13,787✔
120
        logical_start: u32,
13,787✔
121
        count: u32,
13,787✔
122
        epoch: Epoch,
13,787✔
123
    ) -> Result<Self, CoreError> {
13,787✔
124
        if count == 0 {
13,787✔
125
            return Err(CoreError::InvalidCount(0));
1✔
126
        }
13,786✔
127
        let last_logical =
13,785✔
128
            logical_start
13,786✔
129
                .checked_add(count - 1)
13,786✔
130
                .ok_or(CoreError::LogicalRangeOutOfRange {
13,786✔
131
                    logical_start,
13,786✔
132
                    count,
13,786✔
133
                })?;
13,786✔
134
        Timestamp::try_pack(physical_ms, last_logical).map_err(|err| match err {
13,785✔
135
            crate::TimestampError::PhysicalMsOutOfRange { physical_ms, .. } => {
1✔
136
                CoreError::PhysicalMsOutOfRange(physical_ms)
1✔
137
            }
138
            crate::TimestampError::LogicalOutOfRange { .. } => CoreError::LogicalRangeOutOfRange {
1✔
139
                logical_start,
1✔
140
                count,
1✔
141
            },
1✔
142
        })?;
2✔
143
        Ok(WindowGrant {
13,783✔
144
            physical_ms,
13,783✔
145
            logical_start,
13,783✔
146
            count,
13,783✔
147
            epoch,
13,783✔
148
        })
13,783✔
149
    }
13,787✔
150

151
    pub fn physical_ms(&self) -> u64 {
10,141✔
152
        self.physical_ms
10,141✔
153
    }
10,141✔
154
    pub fn logical_start(&self) -> u32 {
10,141✔
155
        self.logical_start
10,141✔
156
    }
10,141✔
157
    pub fn count(&self) -> u32 {
20,281✔
158
        self.count
20,281✔
159
    }
20,281✔
160
    pub fn epoch(&self) -> Epoch {
10,142✔
161
        self.epoch
10,142✔
162
    }
10,142✔
163

164
    /// The first timestamp in the grant. Infallible: [`try_new`](Self::try_new)
165
    /// validated `(physical_ms, logical_start)` is in range, so `pack` cannot
166
    /// trip its `assert!`.
167
    pub fn first(&self) -> Timestamp {
3,629✔
168
        Timestamp::pack(self.physical_ms, self.logical_start)
3,629✔
169
    }
3,629✔
170
    /// The last timestamp in the grant. Infallible: [`try_new`](Self::try_new)
171
    /// validated `(physical_ms, logical_start + count - 1)` is in range (and
172
    /// `count >= 1`, so the subtraction cannot underflow), so `pack` cannot
173
    /// trip its `assert!`.
174
    pub fn last(&self) -> Timestamp {
3,629✔
175
        Timestamp::pack(self.physical_ms, self.logical_start + self.count - 1)
3,629✔
176
    }
3,629✔
177
}
178

179
#[derive(Debug, thiserror::Error, PartialEq, Eq)]
180
pub enum CoreError {
181
    #[error("not leader")]
182
    NotLeader,
183
    #[error("window exhausted; caller must extend before retrying")]
184
    WindowExhausted,
185
    #[error("invalid count: {0}")]
186
    InvalidCount(u32),
187
    #[error("physical_ms {0} exceeds 46-bit maximum")]
188
    PhysicalMsOutOfRange(u64),
189
    #[error("logical range [{logical_start}, +{count}) exceeds the 18-bit logical field")]
190
    LogicalRangeOutOfRange { logical_start: u32, count: u32 },
191
    #[error(
192
        "invalid leadership window: fence_floor {fence_floor} exceeds committed_ceiling {committed_ceiling}"
193
    )]
194
    InvalidLeadershipWindow {
195
        fence_floor: u64,
196
        committed_ceiling: u64,
197
    },
198
    #[error(
199
        "window extension overflow: max(floor {floor}, now_ms {now_ms}) + ahead_ms {ahead_ms} exceeds u64::MAX"
200
    )]
201
    WindowExtensionOverflow {
202
        floor: u64,
203
        now_ms: u64,
204
        ahead_ms: u64,
205
    },
206
}
207

208
/// The result of a `try_commit_window_extension` that passed range validation.
209
///
210
/// A commit either raises the durable bound or is dropped for one of three
211
/// benign, expected reasons. Collapsing both into `Ok(())` left a caller that
212
/// just paid for a `persist_high_water` round-trip unable to tell "I raised the
213
/// bound" from "I silently dropped your durably-persisted value." This type
214
/// preserves the distinction so the server can log/metric the dropped commits —
215
/// a leading indicator of epoch churn or persist reordering.
216
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
217
pub enum CommitOutcome {
218
    /// The durable bound advanced to this value.
219
    Applied(u64),
220
    /// The bound did not move; see [`IgnoreReason`] for why.
221
    Ignored(IgnoreReason),
222
}
223

224
/// Why a [`CommitOutcome::Ignored`] commit left the durable bound unchanged.
225
///
226
/// All three are benign and expected under normal failover: the epoch-fencing
227
/// design of `try_commit_window_extension` deliberately drops late commits from
228
/// a superseded epoch rather than erroring, and the monotonic bound rejects a
229
/// value that does not advance. They are kept apart so a caller can distinguish
230
/// epoch churn ([`NotLeader`](Self::NotLeader) / [`EpochMismatch`](Self::EpochMismatch))
231
/// from persist reordering ([`NotAdvanced`](Self::NotAdvanced)).
232
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
233
pub enum IgnoreReason {
234
    /// The allocator is no longer a leader, so the commit has no window to raise.
235
    NotLeader,
236
    /// The allocator leads a different epoch than the commit targeted; the
237
    /// commit is a late persist from a superseded epoch, fenced out.
238
    EpochMismatch { expected: Epoch, current: Epoch },
239
    /// The allocator still leads the targeted epoch, but the persisted value did
240
    /// not exceed the current bound, so the monotonic bound rejects it.
241
    NotAdvanced { persisted: u64, committed: u64 },
242
}
243

244
#[derive(Debug)]
245
enum State {
246
    NotLeader,
247
    Leader {
248
        epoch: Epoch,
249
        /// Persisted upper bound: the allocator will not issue any timestamp with
250
        /// `physical_ms` greater than this without a fresh `try_commit_window_extension`.
251
        committed_high_water: u64,
252
        /// Next `physical_ms` we are willing to issue at. Initialized to
253
        /// `fence_floor` on leadership gain, then advances monotonically — never
254
        /// retreats below the fence even when `now_ms` is a past value.
255
        next_physical_ms: u64,
256
        /// Next logical counter within `next_physical_ms`.
257
        next_logical: u32,
258
    },
259
}
260

261
pub struct Allocator {
262
    state: State,
263
}
264

265
impl Allocator {
266
    pub fn new() -> Self {
420✔
267
        Allocator {
420✔
268
            state: State::NotLeader,
420✔
269
        }
420✔
270
    }
420✔
271

272
    /// Seed the allocator once the failover fence has durably persisted both
273
    /// the floor and the pre-extended ceiling.
274
    ///
275
    /// `fence_floor` is the first `physical_ms` the new leader may issue —
276
    /// the server sets it to `prior_high_water + 1` so the new leader's
277
    /// timestamps are strictly above any the prior leader could have issued.
278
    ///
279
    /// `committed_ceiling` is the pre-extended upper bound the server has
280
    /// already persisted (typically `fence_floor + window_ms`). It must
281
    /// satisfy `committed_ceiling >= fence_floor` so the allocator can serve
282
    /// `try_grant` immediately without an additional extension round-trip.
283
    pub fn try_on_leadership_gained(
8,390✔
284
        &mut self,
8,390✔
285
        fence_floor: PhysicalMs,
8,390✔
286
        committed_ceiling: PhysicalMs,
8,390✔
287
        epoch: Epoch,
8,390✔
288
    ) -> Result<(), CoreError> {
8,390✔
289
        if committed_ceiling < fence_floor {
8,390✔
290
            return Err(CoreError::InvalidLeadershipWindow {
1✔
291
                fence_floor: fence_floor.get(),
1✔
292
                committed_ceiling: committed_ceiling.get(),
1✔
293
            });
1✔
294
        }
8,389✔
295
        self.state = State::Leader {
8,389✔
296
            epoch,
8,389✔
297
            committed_high_water: committed_ceiling.get(),
8,389✔
298
            next_physical_ms: fence_floor.get(),
8,389✔
299
            next_logical: 0,
8,389✔
300
        };
8,389✔
301
        Ok(())
8,389✔
302
    }
8,390✔
303

304
    pub fn on_leadership_lost(&mut self) {
8,386✔
305
        self.state = State::NotLeader;
8,386✔
306
    }
8,386✔
307

308
    pub fn is_leader(&self) -> bool {
5✔
309
        matches!(self.state, State::Leader { .. })
5✔
310
    }
5✔
311

312
    pub fn epoch(&self) -> Option<Epoch> {
8,184✔
313
        match self.state {
8,184✔
314
            State::Leader { epoch, .. } => Some(epoch),
4,215✔
315
            State::NotLeader => None,
3,969✔
316
        }
317
    }
8,184✔
318

319
    /// Current committed high-water in physical-millisecond units, or `None`
320
    /// when not the leader. The high-water is the upper bound the allocator
321
    /// will not exceed without a fresh `try_commit_window_extension`.
322
    pub fn committed_high_water(&self) -> Option<u64> {
7✔
323
        match self.state {
7✔
324
            State::Leader {
325
                committed_high_water,
4✔
326
                ..
327
            } => Some(committed_high_water),
4✔
328
            State::NotLeader => None,
3✔
329
        }
330
    }
7✔
331

332
    /// Shared count guard for `try_grant` and `would_grant`, kept here so the
333
    /// two entry points cannot drift. The server's extension single-flight
334
    /// relies on `would_grant(now_ms, count) == true` implying the retry
335
    /// `try_grant(now_ms, count)` succeeds (see `service::extend_window`), so
336
    /// the set of rejected counts must be identical on both paths — splitting
337
    /// this check across both methods would let a future edit to one degrade
338
    /// the recheck into a spurious consensus round-trip (or worse).
339
    ///
340
    /// `count == 0` reuses the oversized case's `InvalidCount(count)`; since
341
    /// `count` is `0` there, the surfaced value matches the prior explicit
342
    /// `InvalidCount(0)`.
343
    fn validate_count(count: u32) -> Result<(), CoreError> {
18,636✔
344
        if count == 0 || count > LOGICAL_MAX + 1 {
18,636✔
345
            return Err(CoreError::InvalidCount(count));
4✔
346
        }
18,632✔
347
        Ok(())
18,632✔
348
    }
18,636✔
349

350
    /// Single source of truth for the window-advance simulation and its bounds
351
    /// checks, shared by `try_grant` and `would_grant`. Pure: it takes the
352
    /// relevant state fields by value and mutates nothing, so a failed
353
    /// projection cannot leave allocator state advanced.
354
    ///
355
    /// On success returns the `(physical_ms, logical_start)` the grant would
356
    /// occupy. The two failure variants are kept distinct so `try_grant` can
357
    /// surface the precise error its callers (and tests) expect;
358
    /// `would_grant` collapses both to `false` via `.is_ok()`.
359
    fn project_grant(
14,562✔
360
        next_physical_ms: u64,
14,562✔
361
        next_logical: u32,
14,562✔
362
        committed_high_water: u64,
14,562✔
363
        now_ms: u64,
14,562✔
364
        count: u32,
14,562✔
365
    ) -> Result<(u64, u32), CoreError> {
14,562✔
366
        let mut physical_ms = next_physical_ms;
14,562✔
367
        let mut logical = next_logical;
14,562✔
368

369
        // Advance physical_ms toward wall clock if ahead. next_physical_ms is
370
        // already at or above fence_floor, so a low now_ms simply leaves it there.
371
        if now_ms > physical_ms {
14,562✔
372
            physical_ms = now_ms;
8,979✔
373
            logical = 0;
8,979✔
374
        }
8,979✔
375

376
        // If the current physical_ms cannot fit the request in its remaining
377
        // logical range, advance to the next physical_ms.
378
        if logical as u64 + count as u64 > LOGICAL_MAX as u64 + 1 {
14,562✔
379
            physical_ms += 1;
×
380
            logical = 0;
×
381
        }
14,562✔
382

383
        if physical_ms > PHYSICAL_MS_MAX {
14,562✔
384
            return Err(CoreError::PhysicalMsOutOfRange(physical_ms));
2✔
385
        }
14,560✔
386

387
        // The fence: never issue a timestamp at a physical_ms above the committed
388
        // high-water. If we are at or past the bound, the caller must extend.
389
        if physical_ms > committed_high_water {
14,560✔
390
            return Err(CoreError::WindowExhausted);
731✔
391
        }
13,829✔
392

393
        Ok((physical_ms, logical))
13,829✔
394
    }
14,562✔
395

396
    /// Normalize the post-grant cursor into a packable `(physical_ms, logical)`
397
    /// pair. `project_grant` admits a grant that exactly fills the millisecond,
398
    /// so `logical_start + count` can reach `LOGICAL_MAX + 1` — a logical the
399
    /// packed layout cannot hold. Roll that exact-fill case to the next
400
    /// millisecond at logical 0: this is precisely the position the next
401
    /// `project_grant` call would compute, so behavior is unchanged; the stored
402
    /// state just no longer depends on the implicit "next call always wraps or
403
    /// resets" invariant. (The returned `physical_ms` may equal
404
    /// `PHYSICAL_MS_MAX + 1`, which is never packed directly and is rejected as
405
    /// out-of-range by the next grant's `project_grant`.)
406
    ///
407
    /// Caller contract: `logical_start + count <= LOGICAL_MAX + 1`, which
408
    /// `project_grant`'s logical-range bound already enforces for `try_grant`.
409
    fn advance_cursor(physical_ms: u64, logical_start: u32, count: u32) -> (u64, u32) {
13,781✔
410
        let next_logical = logical_start + count;
13,781✔
411
        if next_logical > LOGICAL_MAX {
13,781✔
412
            (physical_ms + 1, 0)
6✔
413
        } else {
414
            (physical_ms, next_logical)
13,775✔
415
        }
416
    }
13,781✔
417

418
    /// Hot path. Issue `count` timestamps from the in-memory window.
419
    ///
420
    /// Returns `WindowExhausted` when the in-memory remainder cannot cover the request;
421
    /// the caller (typically the server) then runs prepare → persist → commit and retries.
422
    ///
423
    /// State is written back only on success: a failed grant (out-of-range or
424
    /// exhausted window) leaves `next_physical_ms`/`next_logical` untouched.
425
    pub fn try_grant(&mut self, now_ms: u64, count: u32) -> Result<WindowGrant, CoreError> {
18,557✔
426
        Self::validate_count(count)?;
18,557✔
427
        let State::Leader {
428
            epoch,
14,487✔
429
            committed_high_water,
14,487✔
430
            next_physical_ms,
14,487✔
431
            next_logical,
14,487✔
432
        } = &mut self.state
18,555✔
433
        else {
434
            return Err(CoreError::NotLeader);
4,068✔
435
        };
436

437
        let (physical_ms, logical_start) = Self::project_grant(
14,487✔
438
            *next_physical_ms,
14,487✔
439
            *next_logical,
14,487✔
440
            *committed_high_water,
14,487✔
441
            now_ms,
14,487✔
442
            count,
14,487✔
443
        )?;
706✔
444

445
        // project_grant already guarantees physical_ms and the logical range
446
        // are in bounds, so this construction never fails in practice — but
447
        // routing through the checked constructor propagates a CoreError rather
448
        // than panicking should that invariant ever be violated, and keeps this
449
        // path free of the unwrap/expect the crate's panic policy bans.
450
        let grant = WindowGrant::try_new(physical_ms, logical_start, count, *epoch)?;
13,781✔
451
        (*next_physical_ms, *next_logical) =
13,781✔
452
            Self::advance_cursor(physical_ms, logical_start, count);
13,781✔
453
        Ok(grant)
13,781✔
454
    }
18,557✔
455

456
    /// Non-mutating predicate: would `try_grant(now_ms, count)` succeed right
457
    /// now? Used by the server's extension single-flight to decide whether a
458
    /// peer extender has already added enough room, avoiding a redundant
459
    /// `persist_high_water` round-trip. Delegates to the same `project_grant`
460
    /// helper `try_grant` uses, so the exhaustion check cannot drift — a
461
    /// coarser predicate would risk false positives (skip the extension, then
462
    /// fail the outer retry) for requests whose `count` straddles the window edge.
463
    pub fn would_grant(&self, now_ms: u64, count: u32) -> bool {
79✔
464
        if Self::validate_count(count).is_err() {
79✔
465
            return false;
2✔
466
        }
77✔
467
        let State::Leader {
468
            committed_high_water,
75✔
469
            next_physical_ms,
75✔
470
            next_logical,
75✔
471
            ..
472
        } = &self.state
77✔
473
        else {
474
            return false;
2✔
475
        };
476

477
        Self::project_grant(
75✔
478
            *next_physical_ms,
75✔
479
            *next_logical,
75✔
480
            *committed_high_water,
75✔
481
            now_ms,
75✔
482
            count,
75✔
483
        )
75✔
484
        .is_ok()
75✔
485
    }
79✔
486

487
    /// Compute the high-water value the caller should durably persist before
488
    /// calling `try_commit_window_extension`. Does not mutate.
489
    ///
490
    /// Returns `max(committed_high_water + 1, now_ms) + ahead_ms`. The +1 on
491
    /// `committed_high_water` guarantees forward progress when wall clock is
492
    /// behind the persisted bound (rare, but possible after a clock-step-back).
493
    ///
494
    /// Returns `Err(CoreError::NotLeader)` off-leader, matching every other
495
    /// mutating method. A `0` sentinel here would be indistinguishable from a
496
    /// legitimately prepared bound, letting a caller that skipped `is_leader()`
497
    /// proceed as if preparation had succeeded.
498
    pub fn try_prepare_window_extension(
4,220✔
499
        &self,
4,220✔
500
        now_ms: PhysicalMs,
4,220✔
501
        ahead_ms: u64,
4,220✔
502
    ) -> Result<PhysicalMs, CoreError> {
4,220✔
503
        match &self.state {
4,220✔
504
            State::NotLeader => Err(CoreError::NotLeader),
1✔
505
            State::Leader {
506
                committed_high_water,
4,219✔
507
                ..
508
            } => {
509
                debug_assert!(
4,219✔
510
                    *committed_high_water <= PHYSICAL_MS_MAX,
4,219✔
NEW
511
                    "committed_high_water > PHYSICAL_MS_MAX: \
×
NEW
512
                     try_on_leadership_gained / try_commit_window_extension invariant",
×
513
                );
514
                let floor = *committed_high_water + 1;
4,219✔
515
                let now_ms = now_ms.get();
4,219✔
516
                let requested = core::cmp::max(floor, now_ms).checked_add(ahead_ms).ok_or(
4,219✔
517
                    CoreError::WindowExtensionOverflow {
4,219✔
518
                        floor,
4,219✔
519
                        now_ms,
4,219✔
520
                        ahead_ms,
4,219✔
521
                    },
4,219✔
522
                )?;
1✔
523
                // Re-wrap via `PhysicalMs::try_new`: the only remaining bound
524
                // check on this path is on the *derived* sum, no longer on
525
                // each input parameter.
526
                PhysicalMs::try_new(requested)
4,218✔
527
            }
528
        }
529
    }
4,220✔
530

531
    /// Apply a durably-persisted window extension. `persisted_high_water` is
532
    /// the value returned by `ConsensusDriver::persist_high_water`, which is
533
    /// monotonic — it may equal or exceed the value passed to prepare.
534
    ///
535
    /// The `expected_epoch` argument fences out late-arriving commits from a
536
    /// prior leader epoch: if the allocator is no longer at this epoch (either
537
    /// it has lost leadership or a new leader took over), the commit is
538
    /// dropped. Combined with the server's drain barrier, this guarantees a
539
    /// late persist from epoch N cannot raise the durable bound observed by
540
    /// epoch N+M.
541
    ///
542
    /// Returns [`CommitOutcome`]: `Applied` when the bound advanced, or
543
    /// `Ignored` (with the reason) for the three benign drop cases. The 46-bit
544
    /// physical-ceiling invariant on `persisted_high_water` is now enforced by
545
    /// the [`PhysicalMs`] parameter type itself ([`PhysicalMs::try_new`]);
546
    /// the `Result<_, CoreError>` shape is retained for source compatibility
547
    /// with [`try_on_leadership_gained`] / [`try_prepare_window_extension`],
548
    /// so callers can stay uniform under `?`, but no current code path here
549
    /// produces `Err`.
550
    pub fn try_commit_window_extension(
4,216✔
551
        &mut self,
4,216✔
552
        persisted_high_water: PhysicalMs,
4,216✔
553
        expected_epoch: Epoch,
4,216✔
554
    ) -> Result<CommitOutcome, CoreError> {
4,216✔
555
        let persisted_high_water = persisted_high_water.get();
4,216✔
556
        let State::Leader {
557
            epoch,
4,215✔
558
            committed_high_water,
4,215✔
559
            ..
560
        } = &mut self.state
4,216✔
561
        else {
562
            return Ok(CommitOutcome::Ignored(IgnoreReason::NotLeader));
1✔
563
        };
564
        // Epoch fencing takes precedence over the monotonic check: a late
565
        // persist from a superseded epoch must report EpochMismatch even when
566
        // its value also fails to advance, so churn is not masked as reordering.
567
        if *epoch != expected_epoch {
4,215✔
568
            return Ok(CommitOutcome::Ignored(IgnoreReason::EpochMismatch {
1✔
569
                expected: expected_epoch,
1✔
570
                current: *epoch,
1✔
571
            }));
1✔
572
        }
4,214✔
573
        if persisted_high_water <= *committed_high_water {
4,214✔
574
            return Ok(CommitOutcome::Ignored(IgnoreReason::NotAdvanced {
3✔
575
                persisted: persisted_high_water,
3✔
576
                committed: *committed_high_water,
3✔
577
            }));
3✔
578
        }
4,211✔
579
        *committed_high_water = persisted_high_water;
4,211✔
580
        Ok(CommitOutcome::Applied(persisted_high_water))
4,211✔
581
    }
4,216✔
582
}
583

584
impl Default for Allocator {
585
    fn default() -> Self {
1✔
586
        Self::new()
1✔
587
    }
1✔
588
}
589

590
#[cfg(test)]
591
mod tests {
592
    use super::*;
593

594
    // Tiny helper to keep the bound-validated literals readable. Every
595
    // pre-newtype `try_on_leadership_gained(1_000, 5_000, …)` call carried an
596
    // implicit "these literals are within the 46-bit field" precondition;
597
    // wrapping each in `PhysicalMs::try_new(_).unwrap()` would have buried
598
    // every test in unwrap noise without adding coverage (the literals are
599
    // tiny constants under static review). `pms()` keeps the precondition
600
    // explicit at the type level while reading the same as the original.
601
    fn pms(value: u64) -> PhysicalMs {
56✔
602
        PhysicalMs::try_new(value).expect("test literal exceeds PHYSICAL_MS_MAX")
56✔
603
    }
56✔
604

605
    #[test]
606
    fn new_allocator_is_not_leader() {
1✔
607
        let allocator = Allocator::new();
1✔
608
        assert!(!allocator.is_leader());
1✔
609
        assert_eq!(allocator.epoch(), None);
1✔
610
    }
1✔
611

612
    #[test]
613
    fn on_leadership_gained_sets_epoch() {
1✔
614
        let mut allocator = Allocator::new();
1✔
615
        allocator
1✔
616
            .try_on_leadership_gained(pms(1000), pms(5000), Epoch(5))
1✔
617
            .unwrap();
1✔
618
        assert!(allocator.is_leader());
1✔
619
        assert_eq!(allocator.epoch(), Some(Epoch(5)));
1✔
620
    }
1✔
621

622
    #[test]
623
    fn try_on_leadership_gained_rejects_inverted_window() {
1✔
624
        // The per-argument PHYSICAL_MS_MAX checks are now enforced one layer
625
        // out at `PhysicalMs::try_new` (see the `physical_ms` test block below),
626
        // so this method's only remaining error is the cross-argument
627
        // `committed_ceiling < fence_floor` invariant.
628
        let mut allocator = Allocator::new();
1✔
629
        assert_eq!(
1✔
630
            allocator.try_on_leadership_gained(pms(5000), pms(4000), Epoch(5)),
1✔
631
            Err(CoreError::InvalidLeadershipWindow {
632
                fence_floor: 5000,
633
                committed_ceiling: 4000
634
            })
635
        );
636
    }
1✔
637

638
    #[test]
639
    fn on_leadership_lost_clears_state() {
1✔
640
        let mut allocator = Allocator::new();
1✔
641
        allocator
1✔
642
            .try_on_leadership_gained(pms(1000), pms(5000), Epoch(5))
1✔
643
            .unwrap();
1✔
644
        allocator.on_leadership_lost();
1✔
645
        assert!(!allocator.is_leader());
1✔
646
        assert_eq!(allocator.epoch(), None);
1✔
647
    }
1✔
648

649
    #[test]
650
    fn committed_high_water_tracks_leader_state_and_extensions() {
1✔
651
        let mut allocator = Allocator::new();
1✔
652
        assert_eq!(allocator.committed_high_water(), None);
1✔
653

654
        allocator
1✔
655
            .try_on_leadership_gained(pms(1_000), pms(5_000), Epoch(1))
1✔
656
            .unwrap();
1✔
657
        assert_eq!(allocator.committed_high_water(), Some(5_000));
1✔
658

659
        let target = allocator
1✔
660
            .try_prepare_window_extension(pms(2_000), 3_000)
1✔
661
            .unwrap();
1✔
662
        allocator
1✔
663
            .try_commit_window_extension(target, Epoch(1))
1✔
664
            .unwrap();
1✔
665
        assert_eq!(allocator.committed_high_water(), Some(target.get()));
1✔
666

667
        allocator.on_leadership_lost();
1✔
668
        assert_eq!(allocator.committed_high_water(), None);
1✔
669
    }
1✔
670

671
    #[test]
672
    fn try_grant_not_leader() {
1✔
673
        let mut allocator = Allocator::new();
1✔
674
        assert_eq!(allocator.try_grant(1000, 1), Err(CoreError::NotLeader));
1✔
675
    }
1✔
676

677
    #[test]
678
    fn try_grant_zero_count() {
1✔
679
        let mut allocator = Allocator::new();
1✔
680
        allocator
1✔
681
            .try_on_leadership_gained(pms(1000), pms(5000), Epoch(1))
1✔
682
            .unwrap();
1✔
683
        assert_eq!(
1✔
684
            allocator.try_grant(1000, 0),
1✔
685
            Err(CoreError::InvalidCount(0))
686
        );
687
    }
1✔
688

689
    #[test]
690
    fn try_grant_oversized_count() {
1✔
691
        let mut allocator = Allocator::new();
1✔
692
        allocator
1✔
693
            .try_on_leadership_gained(pms(1000), pms(5000), Epoch(1))
1✔
694
            .unwrap();
1✔
695
        let oversized = LOGICAL_MAX + 2;
1✔
696
        assert_eq!(
1✔
697
            allocator.try_grant(1000, oversized),
1✔
698
            Err(CoreError::InvalidCount(oversized))
1✔
699
        );
700
    }
1✔
701

702
    #[test]
703
    fn try_grant_above_committed_is_window_exhausted() {
1✔
704
        // Advancing `now_ms` past `committed_high_water` correctly returns
705
        // WindowExhausted; the server then extends.
706
        let mut allocator = Allocator::new();
1✔
707
        // fence_floor=5_000, ceiling=5_000 (tight window, no pre-extended gap).
708
        allocator
1✔
709
            .try_on_leadership_gained(pms(5_000), pms(5_000), Epoch(1))
1✔
710
            .unwrap();
1✔
711
        // now_ms below floor: clamps to floor=5_000, which equals the ceiling → succeeds.
712
        allocator.try_grant(4_999, 1).unwrap();
1✔
713
        // now_ms above ceiling: window exhausted.
714
        assert_eq!(
1✔
715
            allocator.try_grant(5_001, 1),
1✔
716
            Err(CoreError::WindowExhausted)
717
        );
718
    }
1✔
719

720
    #[test]
721
    fn failed_try_grant_does_not_advance_state() {
1✔
722
        // A grant that fails the exhaustion check must leave the allocator's
723
        // advance state untouched, so a later grant at a lower `now_ms` is not
724
        // pinned to the failed attempt's wall clock.
725
        let mut allocator = Allocator::new();
1✔
726
        // Tight initial window: fence_floor == ceiling == 1_000.
727
        allocator
1✔
728
            .try_on_leadership_gained(pms(1_000), pms(1_000), Epoch(1))
1✔
729
            .unwrap();
1✔
730
        // now_ms far past the ceiling exhausts the window.
731
        assert_eq!(
1✔
732
            allocator.try_grant(5_000, 1),
1✔
733
            Err(CoreError::WindowExhausted)
734
        );
735
        // Extend the durable bound to exactly 2_000.
736
        let target = allocator
1✔
737
            .try_prepare_window_extension(pms(2_000), 0)
1✔
738
            .unwrap();
1✔
739
        assert_eq!(target, pms(2_000)); // max(committed+1=1_001, now=2_000) + 0
1✔
740
        allocator
1✔
741
            .try_commit_window_extension(target, Epoch(1))
1✔
742
            .unwrap();
1✔
743
        // The failed grant must not have pinned next_physical_ms at 5_000: a
744
        // grant at now_ms=2_000 advances cleanly to physical_ms=2_000 (<= the
745
        // committed 2_000). If state had advanced on the failure, next_physical_ms
746
        // would still be 5_000 and this would exhaust the window again.
747
        let grant = allocator.try_grant(2_000, 1).unwrap();
1✔
748
        assert_eq!(grant.physical_ms, 2_000);
1✔
749
        assert_eq!(grant.logical_start, 0);
1✔
750
    }
1✔
751

752
    #[test]
753
    fn try_grant_after_gain_serves_immediately() {
1✔
754
        // The fence has already persisted a pre-extended window, so the allocator
755
        // can serve immediately. Grants start at fence_floor regardless of now_ms.
756
        let mut allocator = Allocator::new();
1✔
757
        allocator
1✔
758
            .try_on_leadership_gained(pms(5_000), pms(10_000), Epoch(1))
1✔
759
            .unwrap();
1✔
760
        let grant = allocator.try_grant(1_000, 1).unwrap();
1✔
761
        // now_ms=1_000 < fence_floor=5_000, so next_physical_ms stays at 5_000.
762
        assert_eq!(grant.physical_ms, 5_000);
1✔
763
        assert_eq!(grant.logical_start, 0);
1✔
764
        assert_eq!(grant.epoch, Epoch(1));
1✔
765
    }
1✔
766

767
    #[test]
768
    fn prepare_window_extension_not_leader() {
1✔
769
        // Off-leader prepare must error like every other mutating method, not
770
        // return a `0` that a caller could mistake for a prepared bound.
771
        let allocator = Allocator::new();
1✔
772
        assert_eq!(
1✔
773
            allocator.try_prepare_window_extension(pms(1000), 3000),
1✔
774
            Err(CoreError::NotLeader)
775
        );
776
    }
1✔
777

778
    #[test]
779
    fn prepare_window_extension_uses_now_ms_when_ahead_of_high_water() {
1✔
780
        let mut allocator = Allocator::new();
1✔
781
        allocator
1✔
782
            .try_on_leadership_gained(pms(1000), pms(1000), Epoch(1))
1✔
783
            .unwrap();
1✔
784
        let target = allocator
1✔
785
            .try_prepare_window_extension(pms(2000), 3000)
1✔
786
            .unwrap();
1✔
787
        assert_eq!(target, pms(5000)); // max(1001, 2000) + 3000
1✔
788
    }
1✔
789

790
    #[test]
791
    fn prepare_window_extension_uses_high_water_floor_when_clock_behind() {
1✔
792
        let mut allocator = Allocator::new();
1✔
793
        allocator
1✔
794
            .try_on_leadership_gained(pms(10_000), pms(10_000), Epoch(1))
1✔
795
            .unwrap();
1✔
796
        let target = allocator
1✔
797
            .try_prepare_window_extension(pms(500), 3000)
1✔
798
            .unwrap();
1✔
799
        // floor = 10_001, clock = 500. max = 10_001. + 3000 = 13_001.
800
        assert_eq!(target, pms(13_001));
1✔
801
    }
1✔
802

803
    #[test]
804
    fn prepare_window_extension_rejects_out_of_range_target() {
1✔
805
        let mut allocator = Allocator::new();
1✔
806
        allocator
1✔
807
            .try_on_leadership_gained(PhysicalMs::MAX, PhysicalMs::MAX, Epoch(1))
1✔
808
            .unwrap();
1✔
809
        assert_eq!(
1✔
810
            allocator.try_prepare_window_extension(PhysicalMs::MAX, 1),
1✔
811
            Err(CoreError::PhysicalMsOutOfRange(PHYSICAL_MS_MAX + 2))
1✔
812
        );
813
    }
1✔
814

815
    #[test]
816
    fn prepare_window_extension_overflow_names_all_operands() {
1✔
817
        // Pre-newtype, the canonical overflow scenario was a saturated clock
818
        // (SystemClock::now_ms saturates to u64::MAX) plus any non-zero
819
        // ahead_ms. The PhysicalMs newtype now rejects that scenario at the
820
        // boundary wrap (PhysicalMs::try_new(u64::MAX) → PhysicalMsOutOfRange),
821
        // surfacing an earlier, more precise error.
822
        //
823
        // The internal overflow path is still reachable, but only via a
824
        // pathologically large `ahead_ms` (a duration, not a physical-ms, so
825
        // it stays an unbounded u64). The error must still name all three
826
        // real operands so the log points at the offending duration, not a
827
        // phantom "someone passed an absurd physical_ms".
828
        let mut allocator = Allocator::new();
1✔
829
        allocator
1✔
830
            .try_on_leadership_gained(pms(1_000), pms(1_000), Epoch(1))
1✔
831
            .unwrap();
1✔
832
        assert_eq!(
1✔
833
            allocator.try_prepare_window_extension(pms(1_000), u64::MAX),
1✔
834
            Err(CoreError::WindowExtensionOverflow {
835
                floor: 1_001,
836
                now_ms: 1_000,
837
                ahead_ms: u64::MAX,
838
            })
839
        );
840
    }
1✔
841

842
    #[test]
843
    fn commit_then_try_grant_succeeds() {
1✔
844
        let mut allocator = Allocator::new();
1✔
845
        allocator
1✔
846
            .try_on_leadership_gained(pms(1000), pms(1000), Epoch(7))
1✔
847
            .unwrap();
1✔
848
        let target = allocator
1✔
849
            .try_prepare_window_extension(pms(1000), 3000)
1✔
850
            .unwrap();
1✔
851
        assert_eq!(
1✔
852
            allocator.try_commit_window_extension(target, Epoch(7)),
1✔
853
            Ok(CommitOutcome::Applied(target.get()))
1✔
854
        );
855
        let grant = allocator.try_grant(1000, 5).unwrap();
1✔
856
        assert_eq!(grant.count, 5);
1✔
857
        assert_eq!(grant.logical_start, 0);
1✔
858
        assert_eq!(grant.epoch, Epoch(7));
1✔
859
        // physical_ms should be at most the persisted high-water.
860
        assert!(grant.physical_ms <= target.get());
1✔
861
    }
1✔
862

863
    #[test]
864
    fn commit_with_lower_value_is_ignored() {
1✔
865
        let mut allocator = Allocator::new();
1✔
866
        allocator
1✔
867
            .try_on_leadership_gained(pms(1000), pms(1000), Epoch(1))
1✔
868
            .unwrap();
1✔
869
        assert_eq!(
1✔
870
            allocator.try_commit_window_extension(pms(5000), Epoch(1)),
1✔
871
            Ok(CommitOutcome::Applied(5000))
872
        );
873
        // A non-advancing commit reports the values so the caller can tell a
874
        // monotonic-bound regression (persist reordering) from epoch churn.
875
        assert_eq!(
1✔
876
            allocator.try_commit_window_extension(pms(3000), Epoch(1)),
1✔
877
            Ok(CommitOutcome::Ignored(IgnoreReason::NotAdvanced {
878
                persisted: 3000,
879
                committed: 5000,
880
            }))
881
        );
882
        // try_grant up to physical_ms=5000 should still work.
883
        let grant = allocator.try_grant(4500, 1).unwrap();
1✔
884
        assert_eq!(grant.physical_ms, 4500);
1✔
885
    }
1✔
886

887
    #[test]
888
    fn commit_with_equal_value_is_ignored_not_applied() {
1✔
889
        // persist_high_water is monotonic and may *equal* the prepared bound; an
890
        // equal value moves nothing, so it is Ignored(NotAdvanced), not Applied.
891
        let mut allocator = Allocator::new();
1✔
892
        allocator
1✔
893
            .try_on_leadership_gained(pms(1000), pms(5000), Epoch(1))
1✔
894
            .unwrap();
1✔
895
        assert_eq!(
1✔
896
            allocator.try_commit_window_extension(pms(5000), Epoch(1)),
1✔
897
            Ok(CommitOutcome::Ignored(IgnoreReason::NotAdvanced {
898
                persisted: 5000,
899
                committed: 5000,
900
            }))
901
        );
902
    }
1✔
903

904
    // (`commit_rejects_out_of_range_high_water` migrated to the `physical_ms`
905
    // test block: the bound check is now at `PhysicalMs::try_new`, so the
906
    // bad value can no longer reach `try_commit_window_extension`.)
907

908
    #[test]
909
    fn try_grant_rejects_out_of_range_clock() {
1✔
910
        let mut allocator = Allocator::new();
1✔
911
        allocator
1✔
912
            .try_on_leadership_gained(pms(1000), PhysicalMs::MAX, Epoch(1))
1✔
913
            .unwrap();
1✔
914
        assert_eq!(
1✔
915
            allocator.try_grant(PHYSICAL_MS_MAX + 1, 1),
1✔
916
            Err(CoreError::PhysicalMsOutOfRange(PHYSICAL_MS_MAX + 1))
1✔
917
        );
918
    }
1✔
919

920
    #[test]
921
    fn commit_at_wrong_epoch_is_silently_dropped() {
1✔
922
        let mut allocator = Allocator::new();
1✔
923
        // fence_floor=1000, ceiling=1000: tight initial window.
924
        allocator
1✔
925
            .try_on_leadership_gained(pms(1000), pms(1000), Epoch(5))
1✔
926
            .unwrap();
1✔
927
        // A late persist from epoch 4 (the prior leader) — fenced out. The
928
        // outcome names both epochs so the caller can metric epoch churn.
929
        assert_eq!(
1✔
930
            allocator.try_commit_window_extension(pms(9_999), Epoch(4)),
1✔
931
            Ok(CommitOutcome::Ignored(IgnoreReason::EpochMismatch {
932
                expected: Epoch(4),
933
                current: Epoch(5),
934
            }))
935
        );
936
        // The allocator's bound did not move; a grant at now=900 clamps to
937
        // floor=1000, and a request with now=1_100 exhausts the window.
938
        allocator.try_grant(900, 1).unwrap();
1✔
939
        assert_eq!(
1✔
940
            allocator.try_grant(1_100, 1),
1✔
941
            Err(CoreError::WindowExhausted)
942
        );
943
    }
1✔
944

945
    #[test]
946
    fn commit_after_leadership_lost_is_ignored() {
1✔
947
        let mut allocator = Allocator::new();
1✔
948
        allocator
1✔
949
            .try_on_leadership_gained(pms(1000), pms(5000), Epoch(1))
1✔
950
            .unwrap();
1✔
951
        allocator.on_leadership_lost();
1✔
952
        assert_eq!(
1✔
953
            allocator.try_commit_window_extension(pms(9_999), Epoch(1)),
1✔
954
            Ok(CommitOutcome::Ignored(IgnoreReason::NotLeader))
955
        );
956
        assert!(!allocator.is_leader());
1✔
957
    }
1✔
958

959
    #[test]
960
    fn would_grant_matches_try_grant_outcome() {
1✔
961
        let mut allocator = Allocator::new();
1✔
962
        // Not leader: never grants.
963
        assert!(!allocator.would_grant(1_000, 1));
1✔
964
        // Invalid counts: never grants.
965
        allocator
1✔
966
            .try_on_leadership_gained(pms(1_000), pms(5_000), Epoch(1))
1✔
967
            .unwrap();
1✔
968
        assert!(!allocator.would_grant(1_000, 0));
1✔
969
        assert!(!allocator.would_grant(1_000, LOGICAL_MAX + 2));
1✔
970
        // Within-window: matches try_grant. now_ms below floor still grants
971
        // (clamped to floor=1_000, ceiling=5_000).
972
        assert!(allocator.would_grant(0, 1));
1✔
973
        // now_ms above ceiling: predicate refuses (would exhaust).
974
        assert!(!allocator.would_grant(5_001, 1));
1✔
975
        // Mid-window now_ms advances the predicate's internal physical_ms.
976
        assert!(allocator.would_grant(2_500, 1));
1✔
977
    }
1✔
978

979
    #[test]
980
    fn would_grant_predicts_logical_wrap_advance() {
1✔
981
        // When (logical + count) overflows the per-ms logical range, the
982
        // predicate (like try_grant) advances physical_ms by 1. If that
983
        // advance leaves the window, would_grant must return false.
984
        let mut allocator = Allocator::new();
1✔
985
        allocator
1✔
986
            .try_on_leadership_gained(pms(1_000), pms(1_000), Epoch(1))
1✔
987
            .unwrap();
1✔
988
        // count >= LOGICAL_MAX + 1 forces the advance branch on a fresh
989
        // window: logical(0) + count(LOGICAL_MAX+1) doesn't overflow on its
990
        // own, but anything one bigger does. Use LOGICAL_MAX + 1 to land at
991
        // the edge, then any non-zero issue advances physical_ms.
992
        allocator.try_grant(1_000, LOGICAL_MAX + 1).unwrap();
1✔
993
        // Next grant of size 1 would advance to physical_ms = 1_001, which
994
        // exceeds the committed ceiling of 1_000.
995
        assert!(!allocator.would_grant(1_000, 1));
1✔
996
    }
1✔
997

998
    #[test]
999
    fn would_grant_returns_false_when_advance_exceeds_physical_max() {
1✔
1000
        // Construct an allocator at PHYSICAL_MS_MAX so the +1 advance
1001
        // crosses the 46-bit ceiling and the predicate refuses.
1002
        let mut allocator = Allocator::new();
1✔
1003
        allocator
1✔
1004
            .try_on_leadership_gained(PhysicalMs::MAX, PhysicalMs::MAX, Epoch(1))
1✔
1005
            .unwrap();
1✔
1006
        // Fill the logical range so the next would_grant call has to
1007
        // advance physical_ms.
1008
        allocator
1✔
1009
            .try_grant(PHYSICAL_MS_MAX, LOGICAL_MAX + 1)
1✔
1010
            .unwrap();
1✔
1011
        assert!(!allocator.would_grant(PHYSICAL_MS_MAX, 1));
1✔
1012
    }
1✔
1013

1014
    #[test]
1015
    fn default_constructs_not_leader_allocator() {
1✔
1016
        let allocator = Allocator::default();
1✔
1017
        assert!(!allocator.is_leader());
1✔
1018
        assert_eq!(allocator.epoch(), None);
1✔
1019
    }
1✔
1020

1021
    #[test]
1022
    fn logical_wraps_to_next_physical_ms() {
1✔
1023
        let mut allocator = Allocator::new();
1✔
1024
        // fence_floor=0, ceiling=0; extend to 10 before granting.
1025
        allocator
1✔
1026
            .try_on_leadership_gained(PhysicalMs::ZERO, PhysicalMs::ZERO, Epoch(1))
1✔
1027
            .unwrap();
1✔
1028
        allocator
1✔
1029
            .try_commit_window_extension(pms(10), Epoch(1))
1✔
1030
            .unwrap();
1✔
1031
        // Issue LOGICAL_MAX+1 logicals at physical_ms=1, then one more should bump to 2.
1032
        let first = allocator.try_grant(1, LOGICAL_MAX + 1).unwrap();
1✔
1033
        assert_eq!(first.physical_ms, 1);
1✔
1034
        assert_eq!(first.logical_start, 0);
1✔
1035
        let second = allocator.try_grant(1, 1).unwrap();
1✔
1036
        assert_eq!(second.physical_ms, 2);
1✔
1037
        assert_eq!(second.logical_start, 0);
1✔
1038
    }
1✔
1039

1040
    #[test]
1041
    fn exact_fill_grant_normalizes_stored_state_to_packable() {
1✔
1042
        // A grant that consumes a millisecond's entire logical range
1043
        // (logical_start + count == LOGICAL_MAX + 1) must not leave the
1044
        // LOGICAL_MAX+1 sentinel in next_logical: that value cannot be packed
1045
        // (Timestamp::pack asserts logical <= LOGICAL_MAX), so the stored state
1046
        // would only be safe by the implicit "next call always wraps" invariant.
1047
        // The write-back normalizes it to the already-rolled position
1048
        // (physical_ms + 1, 0), so stored state is always directly packable.
1049
        let mut allocator = Allocator::new();
1✔
1050
        allocator
1✔
1051
            .try_on_leadership_gained(PhysicalMs::ZERO, PhysicalMs::ZERO, Epoch(1))
1✔
1052
            .unwrap();
1✔
1053
        allocator
1✔
1054
            .try_commit_window_extension(pms(10), Epoch(1))
1✔
1055
            .unwrap();
1✔
1056
        // Fill physical_ms=1 exactly: logical [0, LOGICAL_MAX].
1057
        let grant = allocator.try_grant(1, LOGICAL_MAX + 1).unwrap();
1✔
1058
        assert_eq!(grant.physical_ms, 1);
1✔
1059
        assert_eq!(grant.logical_start, 0);
1✔
1060

1061
        let State::Leader {
1062
            next_physical_ms,
1✔
1063
            next_logical,
1✔
1064
            ..
1065
        } = allocator.state
1✔
1066
        else {
1067
            panic!("expected leader state after a successful grant");
×
1068
        };
1069
        // The stored cursor rolled to the next millisecond at logical 0 …
1070
        assert_eq!(next_physical_ms, 2);
1✔
1071
        assert_eq!(next_logical, 0);
1✔
1072
        // … and is, by construction, a packable timestamp.
1073
        assert!(Timestamp::try_pack(next_physical_ms, next_logical).is_ok());
1✔
1074
    }
1✔
1075

1076
    #[test]
1077
    fn try_new_accepts_valid_grant_and_packs_boundaries() {
1✔
1078
        // A checked grant exposes its fields through accessors and its
1079
        // boundary timestamps pack without panicking — first() at
1080
        // logical_start, last() at logical_start + count - 1.
1081
        let grant = WindowGrant::try_new(1_000, 5, 3, Epoch(7)).unwrap();
1✔
1082
        assert_eq!(grant.physical_ms(), 1_000);
1✔
1083
        assert_eq!(grant.logical_start(), 5);
1✔
1084
        assert_eq!(grant.count(), 3);
1✔
1085
        assert_eq!(grant.epoch(), Epoch(7));
1✔
1086
        assert_eq!(grant.first(), Timestamp::pack(1_000, 5));
1✔
1087
        assert_eq!(grant.last(), Timestamp::pack(1_000, 7));
1✔
1088
    }
1✔
1089

1090
    #[test]
1091
    fn try_new_accepts_max_logical_boundary() {
1✔
1092
        // logical_start + count - 1 == LOGICAL_MAX is the widest in-range grant.
1093
        let grant = WindowGrant::try_new(1_000, 0, LOGICAL_MAX + 1, Epoch(1)).unwrap();
1✔
1094
        assert_eq!(grant.last(), Timestamp::pack(1_000, LOGICAL_MAX));
1✔
1095
    }
1✔
1096

1097
    #[test]
1098
    fn try_new_rejects_zero_count() {
1✔
1099
        // count == 0 would underflow logical_start + count - 1 in last().
1100
        assert_eq!(
1✔
1101
            WindowGrant::try_new(1_000, 0, 0, Epoch(1)),
1✔
1102
            Err(CoreError::InvalidCount(0))
1103
        );
1104
    }
1✔
1105

1106
    #[test]
1107
    fn try_new_rejects_out_of_range_physical_ms() {
1✔
1108
        assert_eq!(
1✔
1109
            WindowGrant::try_new(PHYSICAL_MS_MAX + 1, 0, 1, Epoch(1)),
1✔
1110
            Err(CoreError::PhysicalMsOutOfRange(PHYSICAL_MS_MAX + 1))
1✔
1111
        );
1112
    }
1✔
1113

1114
    #[test]
1115
    fn try_new_rejects_logical_range_overflow() {
1✔
1116
        // last logical (logical_start + count - 1) exceeds the 18-bit field.
1117
        assert_eq!(
1✔
1118
            WindowGrant::try_new(1_000, LOGICAL_MAX, 2, Epoch(1)),
1✔
1119
            Err(CoreError::LogicalRangeOutOfRange {
1120
                logical_start: LOGICAL_MAX,
1121
                count: 2
1122
            })
1123
        );
1124
    }
1✔
1125

1126
    #[test]
1127
    fn try_new_rejects_logical_count_u32_overflow() {
1✔
1128
        // logical_start + (count - 1) overflows u32 before any range check.
1129
        assert_eq!(
1✔
1130
            WindowGrant::try_new(1_000, u32::MAX, 2, Epoch(1)),
1✔
1131
            Err(CoreError::LogicalRangeOutOfRange {
1132
                logical_start: u32::MAX,
1133
                count: 2
1134
            })
1135
        );
1136
    }
1✔
1137

1138
    // ----------------------------------------------------------------
1139
    // PhysicalMs newtype: the construction-site bound check that the
1140
    // three Allocator entry points used to re-implement inline. Every
1141
    // assertion below was previously expressed as a runtime check
1142
    // *inside* try_on_leadership_gained, try_prepare_window_extension,
1143
    // or try_commit_window_extension; they now belong to the type.
1144
    // ----------------------------------------------------------------
1145

1146
    #[test]
1147
    fn physical_ms_accepts_zero() {
1✔
1148
        assert_eq!(PhysicalMs::try_new(0).unwrap().get(), 0);
1✔
1149
    }
1✔
1150

1151
    #[test]
1152
    fn physical_ms_accepts_max() {
1✔
1153
        assert_eq!(
1✔
1154
            PhysicalMs::try_new(PHYSICAL_MS_MAX).unwrap().get(),
1✔
1155
            PHYSICAL_MS_MAX,
1156
        );
1157
    }
1✔
1158

1159
    #[test]
1160
    fn physical_ms_rejects_one_past_max() {
1✔
1161
        // The previously-inline checks in try_on_leadership_gained and
1162
        // try_commit_window_extension lived at this exact boundary;
1163
        // they now live here.
1164
        assert_eq!(
1✔
1165
            PhysicalMs::try_new(PHYSICAL_MS_MAX + 1),
1✔
1166
            Err(CoreError::PhysicalMsOutOfRange(PHYSICAL_MS_MAX + 1)),
1✔
1167
        );
1168
    }
1✔
1169

1170
    #[test]
1171
    fn physical_ms_rejects_u64_max() {
1✔
1172
        // The saturated-clock scenario the old `try_prepare_window_extension`
1173
        // overflow test probed by passing u64::MAX is now caught one layer
1174
        // out, at construction.
1175
        assert_eq!(
1✔
1176
            PhysicalMs::try_new(u64::MAX),
1✔
1177
            Err(CoreError::PhysicalMsOutOfRange(u64::MAX)),
1178
        );
1179
    }
1✔
1180

1181
    #[test]
1182
    fn physical_ms_max_const_matches_try_new() {
1✔
1183
        assert_eq!(
1✔
1184
            PhysicalMs::MAX,
1185
            PhysicalMs::try_new(PHYSICAL_MS_MAX).unwrap(),
1✔
1186
        );
1187
    }
1✔
1188

1189
    #[test]
1190
    fn physical_ms_zero_const_matches_try_new_and_default() {
1✔
1191
        // ZERO, Default::default(), and try_new(0) must all coincide so
1192
        // callers can use any spelling without semantic difference.
1193
        assert_eq!(PhysicalMs::ZERO, PhysicalMs::try_new(0).unwrap());
1✔
1194
        assert_eq!(PhysicalMs::default(), PhysicalMs::ZERO);
1✔
1195
    }
1✔
1196

1197
    #[test]
1198
    fn physical_ms_try_from_matches_try_new() {
1✔
1199
        // TryFrom<u64> is required for generic conversion code; it must
1200
        // produce identical Ok/Err to the inherent try_new on every input.
1201
        let good: u64 = 1_234_567;
1✔
1202
        let from_inherent = PhysicalMs::try_new(good).unwrap();
1✔
1203
        let from_trait: PhysicalMs = good.try_into().unwrap();
1✔
1204
        assert_eq!(from_inherent, from_trait);
1✔
1205

1206
        let bad = PHYSICAL_MS_MAX + 1;
1✔
1207
        let bad_inherent = PhysicalMs::try_new(bad);
1✔
1208
        let bad_trait: Result<PhysicalMs, CoreError> = bad.try_into();
1✔
1209
        assert_eq!(bad_inherent, bad_trait);
1✔
1210
    }
1✔
1211

1212
    #[test]
1213
    fn physical_ms_display_passes_through_inner_value() {
1✔
1214
        // Display is a thin passthrough — it must format identically to the
1215
        // underlying u64 so log lines and error messages read the same after
1216
        // the refactor.
1217
        let v: u64 = 4_242_424_242;
1✔
1218
        assert_eq!(
1✔
1219
            format!("{}", PhysicalMs::try_new(v).unwrap()),
1✔
1220
            format!("{v}"),
1✔
1221
        );
1222
    }
1✔
1223

1224
    #[test]
1225
    fn physical_ms_ordering_follows_inner_value() {
1✔
1226
        // The wrapper exposes Ord/PartialOrd so the allocator can compare
1227
        // bounds (e.g. committed_ceiling < fence_floor) without stripping to
1228
        // u64 — this test pins that the derived ordering matches the inner.
1229
        let a = PhysicalMs::try_new(5).unwrap();
1✔
1230
        let b = PhysicalMs::try_new(10).unwrap();
1✔
1231
        assert!(a < b);
1✔
1232
        assert!(b > a);
1✔
1233
        assert!(a <= a);
1✔
1234
    }
1✔
1235

1236
    #[test]
1237
    fn physical_ms_is_copy_and_eq() {
1✔
1238
        // Compile-time witness: if a future edit accidentally drops `Copy`
1239
        // or `Eq`, several allocator call sites that consume the value
1240
        // twice or compare it inside `assert_eq!` would silently break.
1241
        fn assert_copy_eq<T: Copy + Eq>() {}
1✔
1242
        assert_copy_eq::<PhysicalMs>();
1✔
1243
    }
1✔
1244
}
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE TRIAL · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc