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
//! Cross-shard sequence-key scans for [`crate::db::Database`].
//!
//! Split out of `db.rs` (the ≤500-line file law) as a coherent unit: both public
//! scans fan out over the bounded COMMIT-COLLAPSE §6 executor with identical
//! admission-before-materialisation ordering, so they share the file.
use crate::shard::actor::ShardHandle;
use super::helpers::map_shard_error;
use super::{Database, DatabaseError};
impl Database {
/// Collect every stream's `(stream_key, next_seq)` pair across all shards.
///
/// This walks each shard in parallel, scanning its full key range for the
/// per-stream sequence-metadata keys and decoding each one. It is the
/// O(total entries) traversal that backs the `EventStore` `scan` predicate.
pub fn scan_sequence_keys(&self) -> Result<Vec<(Vec<u8>, u64)>, DatabaseError> {
// Only MATERIALISED shards can hold streams: an un-materialised shard has
// never had a write, so it contributes no sequence keys. Scanning the
// materialised set (not `0..shard_count`) is both correct and O(used).
let timeout = self.timeout;
// Admission is acquired inside `submit` BEFORE `materialised_handles`
// runs, so a blocked submitter holds no handles (§6 admission ordering).
let results = self.executor.submit(
|| Ok(self.router.materialised_handles()),
move |handle: ShardHandle| handle.scan_sequences(timeout),
)?;
let mut streams = Vec::new();
for (_, result) in results {
streams.extend(result.map_err(map_shard_error)?);
}
Ok(streams)
}
/// Collect `(stream_key, next_seq)` pairs from ONLY the named shards.
///
/// The scoped counterpart of [`Self::scan_sequence_keys`]: a node that owns a
/// subset of shards enumerates only its own streams (e.g. to recover exactly
/// the workflows whose event streams it serves) without paying for, or
/// surfacing, streams that live on shards another node owns. Each id must be in
/// `0..shard_count`; an out-of-range id is [`DatabaseError::InvalidShardCount`].
///
/// COMMIT-COLLAPSE §7 (§9.7 RULED): a DUPLICATE id is a typed
/// [`DatabaseError::DuplicateShardId`] refusal (it previously scanned the
/// shard twice, silently duplicating output) — nothing is scanned. Together
/// with the in-range check this caps the executor batch at `shard_count` jobs
/// by construction.
pub fn scan_sequence_keys_for_shards(
&self,
shard_ids: &[usize],
) -> Result<Vec<(Vec<u8>, u64)>, DatabaseError> {
// Validate ids as a true SET before admitting or materialising anything.
let mut seen = std::collections::BTreeSet::new();
for &shard_id in shard_ids {
if !seen.insert(shard_id) {
return Err(DatabaseError::DuplicateShardId { shard_id });
}
}
let timeout = self.timeout;
// Materialise each requested shard on demand: a node recovering exactly
// the streams for shards it adopts must open (and WAL-recover) those
// shards even if they were never touched this lifetime. An out-of-range id
// is still rejected as `InvalidShardCount`. Materialisation runs inside
// `submit` AFTER admission (§6 ordering), so a blocked submitter spawns
// and holds nothing.
let results = self.executor.submit(
|| {
let mut handles = Vec::with_capacity(shard_ids.len());
for &shard_id in shard_ids {
handles.push(self.handle_for_shard(shard_id)?);
}
Ok(handles)
},
move |handle: ShardHandle| handle.scan_sequences(timeout),
)?;
let mut streams = Vec::new();
for (_, result) in results {
streams.extend(result.map_err(map_shard_error)?);
}
Ok(streams)
}
}