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
//! When the services start over a durable store, every in-flight claim in the
//! durable delivery dedup namespace is looked up by its key before any publish
//! is judged: a claim left by a crash or by an unresolved publish is completed
//! when its row is in the stream, released when it is not, and held when
//! nothing can say yet. The claims are read from the namespace's in-flight
//! index (`liminal` `durability/dedup/in_flight.rs`), never by walking the
//! namespace; the first start over a store written before the index existed
//! walks it once to build the index. A store created empty by this process
//! holds no claim, so it is not read at all.
use std::path::Path;
use std::sync::Arc;
use super::unresolved::HeldKey;
use super::{
DELIVERY_DEDUP_NAMESPACE, LiminalConnectionServices, database_exists, database_untouched,
durable_data_dir,
};
use crate::ServerError;
use liminal::durability::bridge::block_on;
use liminal::durability::{DedupCache, DurableStore};
/// Whether the durable store the services start over can already hold
/// in-flight claims.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum StoreHistory {
/// The store was created empty for this process and dies with it: no
/// earlier server wrote to it, so it holds no claim to look up.
#[cfg(any(test, feature = "test-support"))]
Fresh,
/// The store is persistent and holds nothing yet: this start creates it,
/// or an earlier start created it and nothing was ever written to it. It
/// holds no claim, and outlives the process.
Created,
/// The store may hold claims an earlier server left: a persistent store
/// that already exists, or one the caller handed in.
Recorded,
}
impl StoreHistory {
/// The history of the store that will be built for `persistence_path`,
/// answered before it is opened: a persistent store holds nothing when its database
/// does not exist yet or holds only what its creation wrote, and otherwise
/// may hold what the servers before this one wrote.
pub(super) fn of(path: &Path) -> Self {
let data_dir = durable_data_dir(path);
if database_exists(&data_dir) && !database_untouched(&data_dir) {
Self::Recorded
} else {
Self::Created
}
}
/// The durable delivery dedup cache over `store`: a store created by this
/// process holds no claim, so its in-flight index starts empty without
/// reading it.
pub(super) fn delivery_dedup(self, store: Arc<dyn DurableStore>) -> DedupCache {
if self == Self::Recorded {
DedupCache::new(store, DELIVERY_DEDUP_NAMESPACE)
} else {
DedupCache::new_empty(store, DELIVERY_DEDUP_NAMESPACE)
}
}
}
impl LiminalConnectionServices {
/// Resolves the durable namespace's in-flight claims, once, at start.
///
/// A store this process created (`Fresh` or `Created`) is not read. It
/// holds nothing, and its index is recorded by the first keyed publish
/// instead of on the boot path, before the listener binds.
///
/// # Errors
///
/// Returns [`ServerError::ListenerAccept`] naming the dedup namespace when
/// its claims cannot be listed. A claim whose stream cannot be read is not
/// an error: it stays held for the channel's next admission.
pub(super) fn resolve_claims_at_start(&self, history: StoreHistory) -> Result<(), ServerError> {
if history != StoreHistory::Recorded {
return Ok(());
}
let listing_failed = |message: String| ServerError::ListenerAccept {
message: format!(
"in-flight claims in dedup namespace '{}' could not be listed: {message}",
self.dedup.namespace()
),
};
let claims = block_on(self.dedup.in_flight_claims())
.map_err(|error| listing_failed(error.to_string()))?
.map_err(|error| listing_failed(error.to_string()))?;
for (idempotency_key, evidence) in claims {
let Some(evidence) = evidence else {
tracing::error!(
idempotency_key = idempotency_key.as_str(),
"in-flight claim cannot be checked"
);
self.held_keys.hold(&idempotency_key, HeldKey::Uncheckable);
continue;
};
let channel = evidence.channel().to_owned();
self.held_keys.hold(
&idempotency_key,
HeldKey::Checkable {
channel: channel.clone(),
from: evidence.positions().to_vec(),
},
);
// A channel not registered at start keeps the claim held; its first
// keyed admission after registration looks the row up.
let Ok(configured) = self.admit_channel(&channel) else {
continue;
};
if let Err(error) =
self.settle_held_with(&configured.handle, &channel, &idempotency_key)
{
tracing::warn!(
channel = channel.as_str(),
idempotency_key = idempotency_key.as_str(),
%error,
"in-flight claim could not be resolved at start; held for the next admission"
);
}
}
Ok(())
}
}