trine_kv/content/
lease_hold.rs

1use super::{
2    Arc, AtomicBool, AtomicU64, CONTENT_LEASE_MAGIC, CONTENT_PHYSICAL_HOLD_MAGIC,
3    ContentDescriptor, ContentId, ContentLeaseId, ContentLeaseOwnerId, ContentPhysicalHoldId,
4    ContentPhysicalHoldKind, ContentPhysicalHoldOwnerId, Db, Digest, Duration, Error, Ordering,
5    Result, Sha256, StorageDomainId, array_at, decode_chunk, decode_content_id, fmt,
6};
7
8/// Result of one content-authority cleanup transaction.
9#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
10pub struct ContentLifecycleMaintenanceReport {
11    pub(crate) scanned: u64,
12    pub(crate) expired_tokens_removed: u64,
13    pub(crate) expired_leases_removed: u64,
14    pub(crate) inactive_holds_removed: u64,
15}
16
17impl ContentLifecycleMaintenanceReport {
18    /// Returns the number of token, lease, and hold records inspected.
19    #[must_use]
20    pub const fn scanned(self) -> u64 {
21        self.scanned
22    }
23
24    /// Returns expired upload-token authority records removed.
25    #[must_use]
26    pub const fn expired_tokens_removed(self) -> u64 {
27        self.expired_tokens_removed
28    }
29
30    /// Returns expired read-lease records removed.
31    #[must_use]
32    pub const fn expired_leases_removed(self) -> u64 {
33        self.expired_leases_removed
34    }
35
36    /// Returns released or expired physical-hold records removed.
37    #[must_use]
38    pub const fn inactive_holds_removed(self) -> u64 {
39        self.inactive_holds_removed
40    }
41}
42
43#[derive(Debug)]
44pub(crate) struct ContentLease {
45    id: ContentLeaseId,
46    owner_id: ContentLeaseOwnerId,
47    expires_at_unix_ms: AtomicU64,
48}
49
50impl ContentLease {
51    pub(crate) const fn new(
52        id: ContentLeaseId,
53        owner_id: ContentLeaseOwnerId,
54        expires_at_unix_ms: u64,
55    ) -> Self {
56        Self {
57            id,
58            owner_id,
59            expires_at_unix_ms: AtomicU64::new(expires_at_unix_ms),
60        }
61    }
62
63    pub(crate) const fn id(&self) -> ContentLeaseId {
64        self.id
65    }
66
67    pub(crate) const fn owner_id(&self) -> ContentLeaseOwnerId {
68        self.owner_id
69    }
70
71    pub(crate) fn expires_at_unix_ms(&self) -> u64 {
72        self.expires_at_unix_ms.load(Ordering::Acquire)
73    }
74
75    pub(crate) fn publish_expiry(&self, expires_at_unix_ms: u64) {
76        self.expires_at_unix_ms
77            .store(expires_at_unix_ms, Ordering::Release);
78    }
79}
80
81#[derive(Debug, Clone, Copy, PartialEq, Eq)]
82pub(crate) struct ContentLeaseRecord {
83    pub(crate) lease_id: ContentLeaseId,
84    pub(crate) owner_id: ContentLeaseOwnerId,
85    pub(crate) storage_domain_id: StorageDomainId,
86    pub(crate) content_id: ContentId,
87    pub(crate) expires_at_unix_ms: u64,
88}
89
90impl ContentLeaseRecord {
91    pub(crate) fn encode(self) -> Vec<u8> {
92        let mut bytes = Vec::with_capacity(8 + 16 + 16 + 16 + 33 + 8);
93        bytes.extend_from_slice(CONTENT_LEASE_MAGIC);
94        bytes.extend_from_slice(&self.lease_id.to_bytes());
95        bytes.extend_from_slice(&self.owner_id.to_bytes());
96        bytes.extend_from_slice(&self.storage_domain_id.to_bytes());
97        bytes.extend_from_slice(&self.content_id.to_bytes());
98        bytes.extend_from_slice(&self.expires_at_unix_ms.to_le_bytes());
99        bytes
100    }
101
102    pub(crate) fn decode(
103        bytes: &[u8],
104        storage_domain_id: StorageDomainId,
105        content_id: ContentId,
106        lease_id: ContentLeaseId,
107    ) -> Result<Self> {
108        const RECORD_LEN: usize = 8 + 16 + 16 + 16 + 33 + 8;
109        if bytes.len() != RECORD_LEN || bytes.get(..8) != Some(CONTENT_LEASE_MAGIC) {
110            return Err(Error::InvalidFormat {
111                message: "invalid content lease record header or length".to_owned(),
112            });
113        }
114        let stored_lease_id =
115            ContentLeaseId::from_bytes(array_at::<16>(bytes, 8, "content lease identity")?)?;
116        let owner_id =
117            ContentLeaseOwnerId::from_bytes(array_at::<16>(bytes, 24, "content lease owner")?);
118        let stored_domain =
119            StorageDomainId::from_bytes(array_at::<16>(bytes, 40, "content lease storage domain")?);
120        let stored_content = decode_content_id(bytes, 56, "content lease content identity")?;
121        let expires_at_unix_ms =
122            u64::from_le_bytes(array_at::<8>(bytes, 89, "content lease expiry")?);
123        if stored_lease_id != lease_id
124            || stored_domain != storage_domain_id
125            || stored_content != content_id
126        {
127            return Err(Error::Corruption {
128                message: "content lease record differs from its protected key".to_owned(),
129            });
130        }
131        Ok(Self {
132            lease_id,
133            owner_id,
134            storage_domain_id,
135            content_id,
136            expires_at_unix_ms,
137        })
138    }
139}
140
141pub(crate) fn content_lease_prefix(
142    storage_domain_id: StorageDomainId,
143    content_id: ContentId,
144) -> Vec<u8> {
145    let mut key = Vec::with_capacity(16 + 33);
146    key.extend_from_slice(&storage_domain_id.to_bytes());
147    key.extend_from_slice(&content_id.to_bytes());
148    key
149}
150
151pub(crate) fn content_lease_key(
152    storage_domain_id: StorageDomainId,
153    content_id: ContentId,
154    lease_id: ContentLeaseId,
155) -> Vec<u8> {
156    let mut key = content_lease_prefix(storage_domain_id, content_id);
157    key.extend_from_slice(&lease_id.to_bytes());
158    key
159}
160
161#[derive(Debug, Clone, Copy, PartialEq, Eq)]
162pub(crate) enum ContentPhysicalHoldRecordState {
163    Active,
164    Released,
165}
166
167#[derive(Debug, Clone, Copy, PartialEq, Eq)]
168pub(crate) struct ContentPhysicalHoldRecord {
169    pub(crate) hold_id: ContentPhysicalHoldId,
170    pub(crate) owner_id: ContentPhysicalHoldOwnerId,
171    pub(crate) storage_domain_id: StorageDomainId,
172    pub(crate) content_id: ContentId,
173    pub(crate) kind: ContentPhysicalHoldKind,
174    pub(crate) expires_at_unix_ms: u64,
175    pub(crate) state: ContentPhysicalHoldRecordState,
176}
177
178impl ContentPhysicalHoldRecord {
179    pub(crate) const fn is_active_at(self, now_unix_ms: u64) -> bool {
180        matches!(self.state, ContentPhysicalHoldRecordState::Active)
181            && (self.expires_at_unix_ms == 0 || now_unix_ms < self.expires_at_unix_ms)
182    }
183
184    pub(crate) const fn is_released(self) -> bool {
185        matches!(self.state, ContentPhysicalHoldRecordState::Released)
186    }
187
188    pub(crate) const fn released(mut self) -> Self {
189        self.state = ContentPhysicalHoldRecordState::Released;
190        self
191    }
192
193    pub(crate) fn encode(self) -> Vec<u8> {
194        let mut bytes = Vec::with_capacity(8 + 16 + 16 + 16 + 33 + 1 + 1 + 8);
195        bytes.extend_from_slice(CONTENT_PHYSICAL_HOLD_MAGIC);
196        bytes.extend_from_slice(&self.hold_id.to_bytes());
197        bytes.extend_from_slice(&self.owner_id.to_bytes());
198        bytes.extend_from_slice(&self.storage_domain_id.to_bytes());
199        bytes.extend_from_slice(&self.content_id.to_bytes());
200        bytes.push(self.kind.tag());
201        bytes.push(match self.state {
202            ContentPhysicalHoldRecordState::Active => 0,
203            ContentPhysicalHoldRecordState::Released => 1,
204        });
205        bytes.extend_from_slice(&self.expires_at_unix_ms.to_le_bytes());
206        bytes
207    }
208
209    pub(crate) fn decode(
210        bytes: &[u8],
211        storage_domain_id: StorageDomainId,
212        content_id: ContentId,
213        hold_id: ContentPhysicalHoldId,
214    ) -> Result<Self> {
215        const RECORD_LEN: usize = 8 + 16 + 16 + 16 + 33 + 1 + 1 + 8;
216        if bytes.len() != RECORD_LEN || bytes.get(..8) != Some(CONTENT_PHYSICAL_HOLD_MAGIC) {
217            return Err(Error::InvalidFormat {
218                message: "invalid content physical-hold record header or length".to_owned(),
219            });
220        }
221        let stored_hold_id = ContentPhysicalHoldId::from_bytes(array_at::<16>(
222            bytes,
223            8,
224            "content physical-hold identity",
225        )?)?;
226        let owner_id = ContentPhysicalHoldOwnerId::from_bytes(array_at::<16>(
227            bytes,
228            24,
229            "content physical-hold owner",
230        )?);
231        let stored_domain = StorageDomainId::from_bytes(array_at::<16>(
232            bytes,
233            40,
234            "content physical-hold storage domain",
235        )?);
236        let stored_content =
237            decode_content_id(bytes, 56, "content physical-hold content identity")?;
238        let kind = ContentPhysicalHoldKind::from_tag(bytes[89])?;
239        let state = match bytes[90] {
240            0 => ContentPhysicalHoldRecordState::Active,
241            1 => ContentPhysicalHoldRecordState::Released,
242            state => {
243                return Err(Error::UnsupportedFormat {
244                    message: format!("unsupported content physical-hold state {state}"),
245                });
246            }
247        };
248        let expires_at_unix_ms =
249            u64::from_le_bytes(array_at::<8>(bytes, 91, "content physical-hold expiry")?);
250        if stored_hold_id != hold_id
251            || stored_domain != storage_domain_id
252            || stored_content != content_id
253        {
254            return Err(Error::Corruption {
255                message: "content physical-hold record differs from its protected key".to_owned(),
256            });
257        }
258        Ok(Self {
259            hold_id,
260            owner_id,
261            storage_domain_id,
262            content_id,
263            kind,
264            expires_at_unix_ms,
265            state,
266        })
267    }
268}
269
270pub(crate) fn content_physical_hold_prefix(
271    storage_domain_id: StorageDomainId,
272    content_id: ContentId,
273) -> Vec<u8> {
274    content_lease_prefix(storage_domain_id, content_id)
275}
276
277pub(crate) fn content_physical_hold_key(
278    storage_domain_id: StorageDomainId,
279    content_id: ContentId,
280    hold_id: ContentPhysicalHoldId,
281) -> Vec<u8> {
282    let mut key = content_physical_hold_prefix(storage_domain_id, content_id);
283    key.extend_from_slice(&hold_id.to_bytes());
284    key
285}
286
287pub(crate) fn current_epoch_millis() -> Result<u64> {
288    crate::platform::now_unix_millis()
289}
290
291pub(crate) fn duration_millis(duration: Duration, name: &str) -> Result<u64> {
292    let millis = u64::try_from(duration.as_millis())
293        .map_err(|_| Error::invalid_options(format!("{name} milliseconds exceed u64::MAX")))?;
294    if millis == 0 {
295        return Err(Error::invalid_options(format!(
296            "{name} must be at least one millisecond"
297        )));
298    }
299    Ok(millis)
300}
301
302#[derive(Debug)]
303struct ContentPhysicalHoldState {
304    expires_at_unix_ms: AtomicU64,
305    released: AtomicBool,
306}
307
308/// Durable reason that one exact content identity must remain physically readable.
309///
310/// Clones share local release and expiry state, while the protected database
311/// record is the cross-process source of truth. Dropping the value performs no
312/// asynchronous I/O. Expiring holds become inert at their deadline; an
313/// until-released hold must be resumed and explicitly released after a crash.
314#[derive(Clone)]
315pub struct ContentPhysicalHold {
316    db: Db,
317    storage_domain_id: StorageDomainId,
318    content_id: ContentId,
319    hold_id: ContentPhysicalHoldId,
320    owner_id: ContentPhysicalHoldOwnerId,
321    kind: ContentPhysicalHoldKind,
322    state: Arc<ContentPhysicalHoldState>,
323}
324
325impl ContentPhysicalHold {
326    pub(crate) fn from_record(db: Db, record: ContentPhysicalHoldRecord) -> Self {
327        Self {
328            db,
329            storage_domain_id: record.storage_domain_id,
330            content_id: record.content_id,
331            hold_id: record.hold_id,
332            owner_id: record.owner_id,
333            kind: record.kind,
334            state: Arc::new(ContentPhysicalHoldState {
335                expires_at_unix_ms: AtomicU64::new(record.expires_at_unix_ms),
336                released: AtomicBool::new(false),
337            }),
338        }
339    }
340
341    /// Returns the exact physical lifecycle domain.
342    #[must_use]
343    pub const fn storage_domain_id(&self) -> StorageDomainId {
344        self.storage_domain_id
345    }
346
347    /// Returns the immutable byte identity protected by this hold.
348    #[must_use]
349    pub const fn content_id(&self) -> ContentId {
350        self.content_id
351    }
352
353    /// Returns the durable hold identity.
354    #[must_use]
355    pub const fn id(&self) -> ContentPhysicalHoldId {
356        self.hold_id
357    }
358
359    /// Returns the operational hold class.
360    #[must_use]
361    pub const fn kind(&self) -> ContentPhysicalHoldKind {
362        self.kind
363    }
364
365    /// Returns the current exclusive deadline, or `None` for explicit release.
366    ///
367    /// All clones observe a successful renewal only after its durable commit.
368    #[must_use]
369    pub fn expires_at_unix_ms(&self) -> Option<u64> {
370        match self.state.expires_at_unix_ms.load(Ordering::Acquire) {
371            0 => None,
372            expires_at_unix_ms => Some(expires_at_unix_ms),
373        }
374    }
375
376    /// Returns whether this process has observed a successful explicit release.
377    ///
378    /// This local convenience flag is not the cross-process source of truth.
379    #[must_use]
380    pub fn is_released(&self) -> bool {
381        self.state.released.load(Ordering::Acquire)
382    }
383
384    /// Renews an unexpired expiring hold for `ttl` from the current time.
385    ///
386    /// The caller must repeat higher-layer authorization first. Renewal cannot
387    /// revive expiry and does not convert an until-released hold.
388    ///
389    /// # Errors
390    ///
391    /// Returns [`Error::ContentPhysicalHoldNotFound`] after release or missing
392    /// durable state, [`Error::ContentPhysicalHoldExpired`] after the deadline,
393    /// [`Error::InvalidOptions`] for an until-released hold or invalid `ttl`,
394    /// [`Error::ContentPhysicalHoldOwnerMismatch`] for wrong authority,
395    /// [`Error::ReadOnly`] when protected state cannot be written, or a
396    /// storage/conflict error when renewal cannot publish.
397    pub async fn renew(&self, ttl: Duration) -> Result<u64> {
398        self.db.renew_content_physical_hold(self, ttl).await
399    }
400
401    /// Durably releases this exact hold idempotently.
402    ///
403    /// All clones observe release after the durable Released tombstone commits.
404    /// A missing record is accepted as an already-completed retry for recovery
405    /// compatibility. Drop does not call this method.
406    ///
407    /// # Errors
408    ///
409    /// Returns [`Error::ContentPhysicalHoldOwnerMismatch`] if the durable owner
410    /// differs, [`Error::ReadOnly`] when protected state cannot be updated, or
411    /// a storage/conflict error if release cannot converge.
412    pub async fn release(&self) -> Result<()> {
413        self.db.release_content_physical_hold(self).await
414    }
415
416    pub(crate) const fn owner_id(&self) -> ContentPhysicalHoldOwnerId {
417        self.owner_id
418    }
419
420    pub(crate) fn publish_expiry(&self, expires_at_unix_ms: u64) {
421        self.state
422            .expires_at_unix_ms
423            .store(expires_at_unix_ms, Ordering::Release);
424    }
425
426    pub(crate) fn publish_released(&self) {
427        self.state.released.store(true, Ordering::Release);
428    }
429}
430
431impl fmt::Debug for ContentPhysicalHold {
432    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
433        formatter
434            .debug_struct("ContentPhysicalHold")
435            .field("storage_domain_id", &self.storage_domain_id)
436            .field("content_id", &self.content_id)
437            .field("hold_id", &self.hold_id)
438            .field("kind", &self.kind)
439            .field("expires_at_unix_ms", &self.expires_at_unix_ms())
440            .field("released", &self.is_released())
441            .finish_non_exhaustive()
442    }
443}
444
445/// Stable read handle for one sealed immutable `ContentObject`.
446///
447/// The handle fixes one descriptor when opened. This prototype has no physical
448/// relocation, so the descriptor's upload identity and chunk boundaries remain
449/// stable for the handle lifetime. A handle opened with
450/// [`Db::open_content_leased`](crate::Db::open_content_leased) additionally
451/// carries one durable short-lived read lease shared by all of its clones.
452#[derive(Debug, Clone)]
453pub struct ContentHandle {
454    db: Db,
455    descriptor: ContentDescriptor,
456    lease: Option<Arc<ContentLease>>,
457}
458
459impl ContentHandle {
460    pub(crate) fn new(db: Db, descriptor: ContentDescriptor) -> Self {
461        Self {
462            db,
463            descriptor,
464            lease: None,
465        }
466    }
467
468    pub(crate) fn with_lease(mut self, lease: ContentLease) -> Self {
469        self.lease = Some(Arc::new(lease));
470        self
471    }
472
473    pub(crate) fn lease(&self) -> Option<&Arc<ContentLease>> {
474        self.lease.as_ref()
475    }
476
477    /// Returns the immutable content identity fixed by this handle.
478    #[must_use]
479    pub const fn content_id(&self) -> ContentId {
480        self.descriptor.content_id
481    }
482
483    /// Returns the storage domain fixed by this handle.
484    #[must_use]
485    pub const fn storage_domain_id(&self) -> StorageDomainId {
486        self.descriptor.storage_domain_id
487    }
488
489    /// Returns the original byte length.
490    #[must_use]
491    pub const fn len(&self) -> u64 {
492        self.descriptor.length
493    }
494
495    /// Returns whether the original byte sequence is empty.
496    #[must_use]
497    pub const fn is_empty(&self) -> bool {
498        self.descriptor.length == 0
499    }
500
501    /// Returns the durable read lease identity, when this handle was opened
502    /// through [`Db::open_content_leased`](crate::Db::open_content_leased).
503    ///
504    /// Ordinary [`Db::open_content`](crate::Db::open_content) handles return
505    /// `None` and are not protected from future physical reclamation.
506    #[must_use]
507    pub fn lease_id(&self) -> Option<ContentLeaseId> {
508        self.lease.as_deref().map(ContentLease::id)
509    }
510
511    /// Returns the current lease deadline as Unix epoch milliseconds.
512    ///
513    /// All clones share this value. `None` identifies an unleased handle.
514    /// Renewal publishes durable state before this local deadline advances.
515    #[must_use]
516    pub fn lease_expires_at_unix_ms(&self) -> Option<u64> {
517        self.lease.as_deref().map(ContentLease::expires_at_unix_ms)
518    }
519
520    /// Renews this handle's durable read lease for `ttl` from the current time.
521    ///
522    /// The caller must repeat its higher-layer authorization before calling.
523    /// Trine KV validates only the opaque owner and exact content identity. All
524    /// handle clones observe the new deadline after durable publication.
525    ///
526    /// # Errors
527    ///
528    /// Returns [`Error::ContentLeaseNotFound`] for an unleased handle or missing
529    /// durable record, [`Error::ContentLeaseExpired`] once the old deadline is
530    /// reached, [`Error::InvalidOptions`] for a sub-millisecond or overflowing
531    /// lifetime, [`Error::ContentQuarantined`] if quarantine won the race after
532    /// the lease expired, or a storage/conflict error if renewal cannot publish.
533    pub async fn renew_lease(&self, ttl: Duration) -> Result<u64> {
534        self.db.renew_content_lease(self, ttl).await
535    }
536
537    fn ensure_lease_active(&self) -> Result<()> {
538        let Some(lease) = self.lease.as_deref() else {
539            return Ok(());
540        };
541        let now = current_epoch_millis()?;
542        let expires_at_unix_ms = lease.expires_at_unix_ms();
543        if now >= expires_at_unix_ms {
544            return Err(Error::ContentLeaseExpired {
545                expired_at_unix_ms: expires_at_unix_ms,
546            });
547        }
548        Ok(())
549    }
550
551    /// Reads a verified byte range using common half-open range semantics.
552    ///
553    /// `start > len()` returns [`Error::ContentRangeOutOfBounds`]. A zero-byte
554    /// request or `start == len()` returns empty bytes. A request extending past
555    /// EOF returns the verified suffix. Only chunks touched by the result are
556    /// read, and each is verified before any of its payload is copied.
557    ///
558    /// # Errors
559    ///
560    /// Returns a typed range error, a storage error, or
561    /// [`Error::ContentDigestMismatch`] / [`Error::Corruption`] if a chunk is
562    /// missing, malformed, or tampered with.
563    pub async fn read_range(&self, start: u64, length: u64) -> Result<Arc<[u8]>> {
564        let _activity = self.db.begin_activity()?;
565        self.ensure_lease_active()?;
566        if start > self.descriptor.length() {
567            return Err(Error::ContentRangeOutOfBounds {
568                start,
569                length: self.descriptor.length(),
570            });
571        }
572        if length == 0 || start == self.descriptor.length() {
573            return Ok(Arc::from([]));
574        }
575        let end = start.saturating_add(length).min(self.descriptor.length());
576        let result_len = usize::try_from(end - start)
577            .map_err(|_| Error::invalid_options("requested content range exceeds usize"))?;
578        let mut result = Vec::with_capacity(result_len);
579        let chunk_bytes = u64::from(self.descriptor.chunk_bytes());
580        let mut position = start;
581        while position < end {
582            self.ensure_lease_active()?;
583            let chunk_index = position / chunk_bytes;
584            let frame = self
585                .db
586                .read_content_chunk(self.descriptor.upload_id(), chunk_index)
587                .await?
588                .ok_or_else(|| Error::Corruption {
589                    message: format!(
590                        "content {} is missing chunk {chunk_index}",
591                        self.descriptor.content_id()
592                    ),
593                })?;
594            let payload = decode_chunk(&frame, self.descriptor.upload_id(), chunk_index)?;
595            let offset = usize::try_from(position % chunk_bytes)
596                .map_err(|_| Error::invalid_options("content chunk offset exceeds usize"))?;
597            let available = payload
598                .len()
599                .checked_sub(offset)
600                .ok_or_else(|| Error::Corruption {
601                    message: format!("content chunk {chunk_index} is shorter than its descriptor"),
602                })?;
603            let remaining = usize::try_from(end - position)
604                .map_err(|_| Error::invalid_options("content range remainder exceeds usize"))?;
605            let take = available.min(remaining);
606            if take == 0 {
607                return Err(Error::Corruption {
608                    message: format!("content chunk {chunk_index} made no read progress"),
609                });
610            }
611            result.extend_from_slice(&payload[offset..offset + take]);
612            position =
613                position
614                    .checked_add(u64::try_from(take).map_err(|_| {
615                        Error::invalid_options("content range progress exceeds u64")
616                    })?)
617                    .ok_or_else(|| Error::invalid_options("content range position overflow"))?;
618        }
619        Ok(Arc::from(result))
620    }
621
622    /// Creates a sequential stream starting at byte zero.
623    #[must_use]
624    pub fn stream(&self) -> ContentStream {
625        ContentStream {
626            handle: self.clone(),
627            position: 0,
628        }
629    }
630
631    /// Recomputes the complete `ContentId` through a bounded-memory stream.
632    ///
633    /// This additionally verifies each chunk frame. Success means both the
634    /// per-chunk digests and the complete original-byte digest match.
635    ///
636    /// # Errors
637    ///
638    /// Returns a storage, format, or digest error at the first invalid chunk.
639    pub async fn verify(&self) -> Result<()> {
640        let _activity = self.db.begin_activity()?;
641        let mut stream = self.stream();
642        let mut hasher = Sha256::new();
643        while let Some(bytes) = stream.next().await? {
644            hasher.update(&bytes);
645        }
646        let actual = ContentId::from_sha256(hasher.finalize().into());
647        if actual != self.descriptor.content_id() {
648            return Err(Error::ContentDigestMismatch {
649                expected: self.descriptor.content_id().to_string(),
650                actual: actual.to_string(),
651            });
652        }
653        Ok(())
654    }
655}
656
657/// Sequential verified reader over one [`ContentHandle`].
658#[derive(Debug, Clone)]
659pub struct ContentStream {
660    handle: ContentHandle,
661    position: u64,
662}
663
664impl ContentStream {
665    /// Returns the next verified chunk, or `None` at EOF.
666    ///
667    /// Each returned value is at most the upload's configured chunk size. The
668    /// stream therefore does not allocate in proportion to total content size.
669    ///
670    /// # Errors
671    ///
672    /// Propagates the same storage, format, and integrity errors as
673    /// [`ContentHandle::read_range`].
674    pub async fn next(&mut self) -> Result<Option<Arc<[u8]>>> {
675        let _activity = self.handle.db.begin_activity()?;
676        if self.position == self.handle.len() {
677            return Ok(None);
678        }
679        let remaining = self.handle.len() - self.position;
680        let length = remaining.min(u64::from(self.handle.descriptor.chunk_bytes()));
681        let bytes = self.handle.read_range(self.position, length).await?;
682        self.position =
683            self.position
684                .checked_add(u64::try_from(bytes.len()).map_err(|_| {
685                    Error::invalid_options("content stream chunk length exceeds u64")
686                })?)
687                .ok_or_else(|| Error::invalid_options("content stream position overflow"))?;
688        Ok(Some(bytes))
689    }
690
691    /// Returns the next unread original-byte offset.
692    #[must_use]
693    pub const fn position(&self) -> u64 {
694        self.position
695    }
696}
trine_kv/content/lease_hold.rs

trine_kv/content/
lease_hold.rs
