git-ents.gitmain
⌘K
foforge

crates/kernel/ents-receive/src/sink.rs

sink.rs147 lines · 5.1 KB · rusthistorycomment on this file
1//! `EventSink`: the sole destination for post-receive matches
2//! (`receive.event-sink`) — null locally, a durable queue hosted.
3//!
4//! This module also carries the two reference implementations named by the
5//! development plan for this phase: [`NullEventSink`] (the null sink the
6//! phase-4 exit criterion runs against) and [`MemoryEventSink`] (an
7//! in-memory, deduplicating sink demonstrating `receive.dedup` and, paired
8//! with [`crate::reconcile`], `receive.reconstructible`).
9
10use std::collections::BTreeSet;
11use std::sync::{Mutex, MutexGuard, PoisonError};
12
13use gix_hash::ObjectId;
14
15use crate::error::Result;
16
17/// The sole destination for post-receive matches (`receive.event-sink`):
18/// `receive` enqueues one `(effect, oid)` obligation per commit that enters
19/// an effect's work set, and never evaluates the effect itself
20/// (`receive.never-blocks`).
21///
22/// # Errors
23///
24/// [`EventSink::enqueue`] fails only when the sink itself cannot durably
25/// record the obligation (queue I/O, hosted). It is never where an effect
26/// runs or where a verdict is judged.
27///
28/// # Examples
29///
30/// A minimal sink that just counts deliveries — enough to see that
31/// `receive` calls `enqueue` at all, without needing the full dedup
32/// bookkeeping [`MemoryEventSink`] provides.
33///
34/// ```
35/// use std::sync::atomic::{AtomicUsize, Ordering};
36///
37/// use ents_receive::EventSink;
38///
39/// #[derive(Default)]
40/// struct Counting(AtomicUsize);
41///
42/// impl EventSink for Counting {
43/// fn enqueue(&self, _effect: &str, _oid: gix_hash::ObjectId) -> ents_receive::Result<()> {
44/// self.0.fetch_add(1, Ordering::Relaxed);
45/// Ok(())
46/// }
47/// }
48///
49/// let sink = Counting::default();
50/// sink.enqueue("unit", gix_hash::ObjectId::null(gix_hash::Kind::Sha1))
51/// .expect("infallible sink");
52/// assert_eq!(sink.0.load(Ordering::Relaxed), 1);
53/// ```
54// @relation(receive.event-sink, receive.never-blocks, scope=file)
55pub trait EventSink: Send + Sync {
56 /// Enqueue re-evaluation of `effect` for `oid`.
57 ///
58 /// Redelivering the same `(effect, oid)` pair MUST be safe to call
59 /// again — the dedup key is exactly this pair (`receive.dedup`), so a
60 /// conforming sink either folds the duplicate itself ([`MemoryEventSink`]
61 /// does) or leaves de-duplication to whatever drains the queue, as long
62 /// as the eventual *outcome* is exactly-once.
63 ///
64 /// # Errors
65 ///
66 /// Only a genuine sink failure (durable-queue I/O); see the trait's
67 /// own doc.
68 fn enqueue(&self, effect: &str, oid: ObjectId) -> Result<()>;
69}
70
71/// The null `EventSink`: drops every obligation.
72///
73/// This is the local deployment's reference sink (`receive.event-sink`:
74/// "null locally") and the one the phase-4 exit criterion runs `receive`
75/// against — a local write path with no effect crate linked yet has nothing
76/// useful to enqueue into.
77///
78/// # Examples
79///
80/// ```
81/// use ents_receive::{EventSink, NullEventSink};
82///
83/// let sink = NullEventSink;
84/// sink.enqueue("unit", gix_hash::ObjectId::null(gix_hash::Kind::Sha1))
85/// .expect("the null sink never fails");
86/// ```
87// @relation(receive.event-sink, scope=file)
88#[derive(Debug, Clone, Copy, Default)]
89pub struct NullEventSink;
90
91impl EventSink for NullEventSink {
92 fn enqueue(&self, _effect: &str, _oid: ObjectId) -> Result<()> {
93 Ok(())
94 }
95}
96
97/// An in-memory, deduplicating `EventSink`: the reference implementation
98/// `receive.dedup` and `receive.reconstructible` describe.
99///
100/// Redelivering the same `(effect, oid)` pair is a no-op — the set, not a
101/// counter, is the state — which is what makes redelivery from an
102/// at-least-once queue yield exactly-once outcomes (`receive.dedup`). This
103/// type MAY lose its state on crash (it is exactly that: in-memory); the
104/// composition root is expected to call [`crate::reconcile`] against
105/// repository state at startup to rebuild it before serving further pushes,
106/// per `receive.reconstructible` — the durable queue this stands in for is a
107/// performance optimization, never a correctness requirement.
108///
109/// # Examples
110///
111/// ```
112/// use ents_receive::{EventSink, MemoryEventSink};
113///
114/// let sink = MemoryEventSink::default();
115/// let oid = gix_hash::ObjectId::null(gix_hash::Kind::Sha1);
116///
117/// sink.enqueue("unit", oid).expect("infallible sink");
118/// sink.enqueue("unit", oid).expect("redelivery is a no-op");
119///
120/// assert_eq!(sink.pending(), vec![("unit".to_owned(), oid)]);
121/// ```
122// @relation(receive.dedup, receive.reconstructible, scope=file)
123#[derive(Debug, Default)]
124pub struct MemoryEventSink {
125 pending: Mutex<BTreeSet<(String, ObjectId)>>,
126}
127
128impl MemoryEventSink {
129 /// Every distinct `(effect, oid)` obligation enqueued so far, in
130 /// sorted order.
131 #[must_use]
132 pub fn pending(&self) -> Vec<(String, ObjectId)> {
133 self.locked().iter().cloned().collect()
134 }
135
136 fn locked(&self) -> MutexGuard<'_, BTreeSet<(String, ObjectId)>> {
137 self.pending.lock().unwrap_or_else(PoisonError::into_inner)
138 }
139}
140
141impl EventSink for MemoryEventSink {
142 // @relation(receive.dedup, scope=function)
143 fn enqueue(&self, effect: &str, oid: ObjectId) -> Result<()> {
144 self.locked().insert((effect.to_owned(), oid));
145 Ok(())
146 }
147}