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
//! [`NodeRangedStore`]: the node-side [`RangedStore`] backend over the
//! iroh-blobs cache (#1621).
//!
//! A thin, per-blob adapter over [`CacheEngine`]'s
//! present/missing/admit/export API, adding no new storage
//! logic. `finalize` relies on the node's own auto-promotion: once every
//! chunk of a blob is present the store already reports it complete, so
//! finalize is just that check, erroring [`RangedStoreError::Incomplete`]
//! otherwise.
use bytes::Bytes;
use decdn_bao_range::{AlignedRange, RangedFuture, RangedStore, RangedStoreError};
use iroh_blobs::Hash;
use crate::engine::CacheEngine;
use crate::serve_store::{EncodeStream, PresentRangeWatch, ServeStore};
/// Node-side [`RangedStore`]: a per-blob view over the iroh-blobs cache.
#[derive(Debug)]
pub struct NodeRangedStore {
engine: CacheEngine,
hash: Hash,
total_bytes: u64,
}
impl NodeRangedStore {
/// Wrap `engine`'s view of `hash` (a `total_bytes`-byte blob) as a
/// [`RangedStore`]. `engine` is cheap to clone (`Arc`-backed), so this
/// holds it by value rather than behind an extra `Arc`.
#[must_use]
pub const fn new(engine: CacheEngine, hash: Hash, total_bytes: u64) -> Self {
Self {
engine,
hash,
total_bytes,
}
}
/// The blob hash this store is scoped to.
#[must_use]
pub const fn hash(&self) -> Hash {
self.hash
}
/// The wrapped [`CacheEngine`] handle. Lets a node-crate wrapper (#1621 Task
/// 10, `NodeAdmitStore`) reach the engine's `admit_bao_stream` without
/// duplicating the `(engine, hash, total_bytes)` triple it already holds.
#[must_use]
pub const fn engine(&self) -> &CacheEngine {
&self.engine
}
/// Does a deny/blacklist/evict/quarantine gate currently refuse this hash? A
/// pure in-memory check ([`CacheEngine::refuses`] reads in-memory sets and
/// never awaits a store hop), so the serve encoder can fold takedown into its per-leaf
/// presence check without a store round trip. Mirrors the guard
/// [`CacheEngine::present_ranges`] applies before it answers presence.
#[must_use]
pub fn refuses(&self) -> bool {
self.engine.refuses(self.hash)
}
}
fn backend<E: std::error::Error + Send + Sync + 'static>(e: E) -> RangedStoreError {
RangedStoreError::Backend(Box::new(e))
}
impl RangedStore for NodeRangedStore {
fn total_bytes(&self) -> u64 {
self.total_bytes
}
fn present_ranges(&self) -> RangedFuture<'_, bao_tree::ChunkRanges> {
Box::pin(async move {
let p = self
.engine
.present_ranges(self.hash)
.await
.map_err(backend)?;
Ok(p.chunk_ranges().clone())
})
}
fn missing_ranges(
&self,
byte_offset: u64,
byte_len: u64,
) -> RangedFuture<'_, bao_tree::ChunkRanges> {
Box::pin(async move {
// Validate + align locally so a start at or past the blob end is a
// typed `Alignment` error, not an opaque backend fault (a bad start is
// an argument error, and the client backend classifies it the same
// way). An end past the blob clamps to it rather than erroring: the caller here is a pull leg driving a request whose
// claimed size can grow, so its gap computation must clamp exactly as
// the serve tiers do, or a signed `ok: true` response then faults
// fetching its own pull range.
let aligned =
decdn_bao_range::align_range_clamped(byte_offset, byte_len, self.total_bytes)?;
let present = self
.engine
.present_ranges(self.hash)
.await
.map_err(backend)?;
Ok(aligned.chunk_ranges().clone() - present.chunk_ranges())
})
}
fn admit(&self, range: AlignedRange, bao_bytes: Bytes) -> RangedFuture<'_, ()> {
Box::pin(async move {
self.engine
.admit_bao(self.hash, range.chunk_ranges().clone(), bao_bytes)
.await
.map_err(backend)
})
}
fn read(&self, byte_offset: u64, byte_len: u64) -> RangedFuture<'_, Bytes> {
Box::pin(async move {
// Reject a start at or past the blob end as a typed `Alignment` error
// before touching the store. An end past the blob clamps instead of
// erroring; `read` is byte-exact, so the clamped content length, not
// the raw, possibly-overshooting `byte_len`, is what gets forwarded
// to the backend.
let aligned =
decdn_bao_range::align_range_clamped(byte_offset, byte_len, self.total_bytes)?;
// `byte_len == 0` keeps its "to end" meaning: `export_range` resolves
// that dynamically against the store's own reported size, which a
// partial blob may not have yet. Only a non-zero, explicit `byte_len`
// clamps to the blob's end here.
let clamped_len = if byte_len == 0 {
0
} else {
aligned
.blob_size()
.saturating_sub(byte_offset)
.min(byte_len)
};
self.engine
.export_range(self.hash, byte_offset, clamped_len)
.await
.map_err(backend)
})
}
fn is_complete(&self) -> RangedFuture<'_, bool> {
Box::pin(async move {
Ok(self
.engine
.present_ranges(self.hash)
.await
.map_err(backend)?
.is_complete())
})
}
fn finalize(&self) -> RangedFuture<'_, ()> {
Box::pin(async move {
if self
.engine
.present_ranges(self.hash)
.await
.map_err(backend)?
.is_complete()
{
Ok(())
} else {
Err(RangedStoreError::Incomplete)
}
})
}
}
impl ServeStore for NodeRangedStore {
fn observe(&self) -> RangedFuture<'_, PresentRangeWatch> {
Box::pin(async move {
self.engine
.observe_present_ranges(self.hash)
.await
.map_err(backend)
})
}
fn encode_range(&self, byte_offset: u64, byte_len: u64) -> RangedFuture<'_, EncodeStream> {
Box::pin(async move {
let stream = self
.engine
.export_bao_range_stream(self.hash, byte_offset, byte_len, self.total_bytes)
.await
.map_err(backend)?;
Ok(Box::pin(futures_util::StreamExt::map(stream, |item| {
item.map_err(backend)
})) as EncodeStream)
})
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used)] // tests
mod tests;