buntrail_engine/remote/
proof.rs

1//! Worker-only proof transport. Events are provisional until observe_owned returns success.
2use super::*;
3use buntrail_protocol::{
4    frame::{Frame, TransferVerifier},
5    types::ProofAction,
6};
7use buntrail_services::git::proof_codec::{self, output::OutputReader};
8pub use buntrail_services::git::{
9    proof_codec::output::{EntryGroup, ObservationEvent, OwnedObservation},
10    worktree_proof::WorktreePreparation,
11};
12use std::{
13    io::{self, Read, Write},
14    time::{Duration, Instant},
15};
16
17fn unavailable(message: &str) -> CoreError {
18    CoreError::new(ErrorCode::Unavailable, message)
19}
20fn response(message: Message) -> CoreError {
21    match message {
22        Message::Error { code, safe_message } => w::core_error(&code, &safe_message),
23        _ => w::invalid(),
24    }
25}
26struct Count(u64);
27impl Write for Count {
28    fn write(&mut self, b: &[u8]) -> io::Result<usize> {
29        self.0 = self
30            .0
31            .checked_add(b.len() as u64)
32            .ok_or_else(|| io::Error::other("proof length overflow"))?;
33        Ok(b.len())
34    }
35    fn flush(&mut self) -> io::Result<()> {
36        Ok(())
37    }
38}
39#[derive(Clone, Copy)]
40enum Part {
41    Lookup,
42    Original,
43    Approvals,
44    Aliases,
45}
46impl Part {
47    fn chunk(self, id: String, offset: u64) -> Message {
48        match self {
49            Self::Lookup => Message::ProofLookupChunk {
50                transfer_id: id,
51                offset,
52            },
53            Self::Original => Message::ProofOriginalChunk {
54                transfer_id: id,
55                offset,
56            },
57            Self::Approvals => Message::ProofApprovalsChunk {
58                transfer_id: id,
59                offset,
60            },
61            Self::Aliases => Message::ProofAliasesChunk {
62                transfer_id: id,
63                offset,
64            },
65        }
66    }
67    fn end(self, id: String, sha256: String) -> Message {
68        match self {
69            Self::Lookup => Message::ProofLookupEnd {
70                transfer_id: id,
71                sha256,
72            },
73            Self::Original => Message::ProofOriginalEnd {
74                transfer_id: id,
75                sha256,
76            },
77            Self::Approvals => Message::ProofApprovalsEnd {
78                transfer_id: id,
79                sha256,
80            },
81            Self::Aliases => Message::ProofAliasesEnd {
82                transfer_id: id,
83                sha256,
84            },
85        }
86    }
87}
88fn next(
89    remote: &RemoteServices,
90    request: &mut transport::Request,
91    deadline: Instant,
92    cancellable: bool,
93) -> Result<Frame, CoreError> {
94    let mut cx = Context::from_waker(std::task::Waker::noop());
95    loop {
96        if let Poll::Ready(frame) = request.poll(&mut cx) {
97            let frame = frame?;
98            if !matches!(frame.header.message, Message::ProofOutputBegin { .. })
99                && frame.header.stream_id != request.header.stream_id
100            {
101                return Err(w::invalid());
102            }
103            return Ok(frame);
104        }
105        if cancellable {
106            remote.check_cancellation()?;
107        }
108        if Instant::now() >= deadline {
109            return Err(CoreError::new(
110                ErrorCode::Unknown,
111                "proof transfer absolute deadline exceeded; inspect original operation without replay",
112            ));
113        }
114        std::thread::park_timeout(Duration::from_millis(2));
115    }
116}
117struct Upload<'a> {
118    remote: &'a RemoteServices,
119    request: &'a mut transport::Request,
120    part: Part,
121    deadline: Instant,
122    bytes: Vec<u8>,
123    offset: u64,
124    sequence: u64,
125    hash: sha2::Sha256,
126}
127impl<'a> Upload<'a> {
128    fn new(
129        remote: &'a RemoteServices,
130        request: &'a mut transport::Request,
131        part: Part,
132        deadline: Instant,
133    ) -> Self {
134        Self {
135            remote,
136            request,
137            part,
138            deadline,
139            bytes: Vec::with_capacity(65536),
140            offset: 0,
141            sequence: 0,
142            hash: sha2::Sha256::new(),
143        }
144    }
145    fn send(&self, message: Message, payload: Vec<u8>) -> Result<(), CoreError> {
146        if matches!(self.part, Part::Lookup) {
147            self.request.send_control(message, payload)
148        } else {
149            self.request.send(message, payload)
150        }
151    }
152    fn drain(&mut self) -> Result<(), CoreError> {
153        if self.bytes.is_empty() {
154            return Ok(());
155        }
156        if !matches!(self.part, Part::Lookup) {
157            self.remote.check_cancellation()?;
158        }
159        let id = self
160            .request
161            .header
162            .stream_id
163            .clone()
164            .ok_or_else(w::invalid)?;
165        self.request.header.sequence = Some(self.sequence);
166        let payload = std::mem::replace(&mut self.bytes, Vec::with_capacity(65536));
167        let length = payload.len();
168        self.hash.update(&payload);
169        self.send(self.part.chunk(id.clone(), self.offset), payload)?;
170        let frame = next(
171            self.remote,
172            self.request,
173            self.deadline,
174            !matches!(self.part, Part::Lookup),
175        )?;
176        match frame.header.message {
177            Message::Credit { bytes }
178                if bytes as usize == length && frame.header.stream_id.as_ref() == Some(&id) => {}
179            m => return Err(response(m)),
180        }
181        self.offset += length as u64;
182        self.sequence = self.sequence.checked_add(1).ok_or_else(w::invalid)?;
183        Ok(())
184    }
185    fn finish(mut self, length: u64) -> Result<(), CoreError> {
186        self.drain()?;
187        if self.offset != length {
188            return Err(w::invalid());
189        }
190        self.request.header.sequence = Some(self.sequence);
191        self.send(
192            self.part.end(
193                self.request
194                    .header
195                    .stream_id
196                    .clone()
197                    .ok_or_else(w::invalid)?,
198                w::hex(&self.hash.clone().finalize()),
199            ),
200            Vec::new(),
201        )?;
202        self.request.header.sequence = None;
203        Ok(())
204    }
205}
206impl Write for Upload<'_> {
207    fn write(&mut self, mut b: &[u8]) -> io::Result<usize> {
208        let total = b.len();
209        while !b.is_empty() {
210            let n = b.len().min(65536 - self.bytes.len());
211            self.bytes.extend_from_slice(&b[..n]);
212            b = &b[n..];
213            if self.bytes.len() == 65536 {
214                self.drain().map_err(io::Error::other)?;
215            }
216        }
217        Ok(total)
218    }
219    fn flush(&mut self) -> io::Result<()> {
220        self.drain().map_err(io::Error::other)
221    }
222}
223fn approvals<'a>(
224    out: &mut impl Write,
225    remote: &RemoteServices,
226    values: impl Iterator<Item = &'a GitSaveApproval>,
227) -> Result<(), CoreError> {
228    for a in values {
229        if a.actor.client != remote.lease.client || a.actor.runtime != remote.lease.runtime {
230            return Err(CoreError::new(
231                ErrorCode::TargetChanged,
232                "save approval client or runtime changed",
233            ));
234        }
235        for b in [
236            a.operation.0.as_slice(),
237            a.request_digest.as_slice(),
238            a.actor.client.0.as_slice(),
239            a.actor.workspace.0.as_slice(),
240            a.actor.lease.0.as_slice(),
241        ] {
242            out.write_all(b).map_err(io_error)?;
243        }
244    }
245    Ok(())
246}
247struct Download<'a> {
248    remote: &'a RemoteServices,
249    request: &'a mut transport::Request,
250    id: String,
251    verifier: TransferVerifier,
252    sequence: u64,
253    buffer: std::io::Cursor<Vec<u8>>,
254    seal: Option<[u8; 32]>,
255    deadline: Instant,
256}
257impl Read for Download<'_> {
258    fn read(&mut self, b: &mut [u8]) -> io::Result<usize> {
259        if b.is_empty() {
260            return Ok(0);
261        }
262        loop {
263            let n = self.buffer.read(b)?;
264            if n != 0 {
265                return Ok(n);
266            }
267            if !self.buffer.get_ref().is_empty() {
268                self.request
269                    .credit(&self.id, self.buffer.get_ref().len())
270                    .map_err(io::Error::other)?;
271                self.buffer = std::io::Cursor::new(Vec::new());
272            }
273            if self.seal.is_some() {
274                return Ok(0);
275            }
276            let frame =
277                next(self.remote, self.request, self.deadline, true).map_err(io::Error::other)?;
278            if frame.header.stream_id.as_ref() != Some(&self.id)
279                || frame.header.sequence != Some(self.sequence)
280            {
281                return Err(io::Error::other("output owner or sequence mismatch"));
282            }
283            self.sequence = self
284                .sequence
285                .checked_add(1)
286                .ok_or_else(|| io::Error::other("output sequence overflow"))?;
287            match frame.header.message {
288                Message::ProofOutputChunk {
289                    transfer_id,
290                    offset,
291                } => {
292                    self.verifier
293                        .chunk(&transfer_id, offset, &frame.payload)
294                        .map_err(io::Error::other)?;
295                    self.buffer = std::io::Cursor::new(frame.payload);
296                }
297                Message::ProofOutputEnd {
298                    transfer_id,
299                    sha256,
300                } => {
301                    let digest = w::digest(&sha256).map_err(io::Error::other)?;
302                    self.verifier
303                        .end(&transfer_id, digest)
304                        .map_err(io::Error::other)?;
305                    self.seal = Some(digest);
306                }
307                m => return Err(io::Error::other(response(m))),
308            }
309        }
310    }
311}
312impl RemoteServices {
313    fn proof_capability(&self) -> Result<(), CoreError> {
314        if !self
315            .negotiated_capabilities()
316            .iter()
317            .any(|c| c == "git.worktree_proof")
318        {
319            return Err(unavailable(
320                "active helper lacks git.worktree_proof; preserve its existing sessions",
321            ));
322        }
323        Ok(())
324    }
325    /// Freeze the complete aliases under the caller's registration barrier. Events remain
326    /// provisional until this method validates the complete codec and authenticated end seal.
327    pub fn observe_owned<'a>(
328        &self,
329        worktree: WorktreeIdentity,
330        aliases: impl Iterator<Item = &'a FileRef> + Clone,
331        mut on_event: impl FnMut(ObservationEvent) -> Result<(), CoreError>,
332    ) -> Result<OwnedObservation, CoreError> {
333        self.proof_capability()?;
334        self.check(&worktree.target)?;
335        if !self.transport.owns_registration(&self.lease) {
336            return Err(unavailable("exact observation registration changed"));
337        }
338        self.check_cancellation()?;
339        let mut count = Count(0);
340        proof_codec::write_aliases(&mut count, &self.target, aliases.clone())?;
341        let id = w::uuid(buntrail_services::connections::local::new_id()?);
342        let deadline = Instant::now() + Duration::from_secs(120);
343        let mut request = self.transport.start(
344            Message::ProofObserveBegin {
345                transfer_id: id.clone(),
346                total_length: count.0,
347                lease_id: w::uuid(self.lease.lease.0),
348                worktree: w::Worktree::from_core(&worktree)?,
349            },
350            None,
351            Some(id),
352        )?;
353        match next(self, &mut request, deadline, true)?.header.message {
354            Message::Ack {} => {}
355            m => return Err(response(m)),
356        }
357        let mut upload = Upload::new(self, &mut request, Part::Aliases, deadline);
358        proof_codec::write_aliases(&mut upload, &self.target, aliases)?;
359        upload.finish(count.0)?;
360        let begin = next(self, &mut request, deadline, true)?;
361        let Message::ProofOutputBegin {
362            transfer_id,
363            total_length,
364        } = begin.header.message
365        else {
366            return Err(response(begin.header.message));
367        };
368        if begin.header.stream_id.as_ref() != Some(&transfer_id) {
369            return Err(w::invalid());
370        }
371        request.header.stream_id = Some(transfer_id.clone());
372        let mut download = Download {
373            remote: self,
374            request: &mut request,
375            id: transfer_id.clone(),
376            verifier: TransferVerifier::new(transfer_id, total_length).map_err(|_| w::invalid())?,
377            sequence: 0,
378            buffer: std::io::Cursor::new(Vec::new()),
379            seal: None,
380            deadline,
381        };
382        let mut output = OutputReader::new(&mut download, total_length, &self.target)?;
383        let observation = output.observation_id();
384        let result = (|| {
385            while let Some(event) = output.next_event()? {
386                on_event(event)?;
387            }
388            let digest = output.wire_digest();
389            let summary = output.finish()?;
390            Ok((summary, digest))
391        })();
392        let result = result.and_then(|(summary, digest)| {
393            if digest.is_none() || digest != download.seal {
394                Err(w::invalid())
395            } else {
396                Ok(summary)
397            }
398        });
399        drop(download);
400        if result.is_ok() {
401            request.finish();
402        }
403        drop(request);
404        if result.is_err() {
405            let _ = self.release_observation(observation);
406        }
407        result
408    }
409    fn lookup_prefix(
410        &self,
411        original: &GitMutationRequest,
412        action: ProofAction,
413    ) -> Result<(transport::Request, Message, u64), CoreError> {
414        self.check(&original.worktree.target)?;
415        let mut count = Count(0);
416        proof_codec::write_original(&mut count, original, &self.target)?;
417        let id = w::uuid(buntrail_services::connections::local::new_id()?);
418        let deadline = Instant::now() + Duration::from_secs(120);
419        let mut request = self.transport.start(
420            Message::ProofLookupBegin {
421                transfer_id: id.clone(),
422                total_length: count.0,
423                action,
424                lease_id: w::uuid(self.lease.lease.0),
425            },
426            Some(original.operation),
427            Some(id),
428        )?;
429        match next(self, &mut request, deadline, false)?.header.message {
430            Message::ProofLookupReady { credit: 65536 } => {}
431            m => return Err(response(m)),
432        }
433        let mut upload = Upload::new(self, &mut request, Part::Lookup, deadline);
434        proof_codec::write_original(&mut upload, original, &self.target)?;
435        upload.finish(count.0)?;
436        let ready = next(self, &mut request, deadline, false)?;
437        Ok((request, ready.header.message, count.0))
438    }
439
440    /// Worker-only existing lookup. This cannot promote a missing original into new work.
441    pub fn existing_original(
442        &self,
443        original: &GitMutationRequest,
444    ) -> Result<Option<OperationReceipt>, CoreError> {
445        if !self
446            .negotiated_capabilities()
447            .iter()
448            .any(|c| c == "git.original_lookup")
449        {
450            return Err(unavailable(
451                "Active helper cannot stream original lookup; preserve the existing supervisor",
452            ));
453        }
454        let (mut request, message, _) = self.lookup_prefix(original, ProofAction::Lookup)?;
455        request.finish();
456        match message {
457            Message::ProofMissing {} => Ok(None),
458            Message::ProofExisting { receipt } => {
459                let receipt = receipt.core(&self.target)?;
460                if receipt.id != original.operation {
461                    return Err(w::invalid());
462                }
463                Ok(Some(receipt))
464            }
465            other => Err(response(other)),
466        }
467    }
468    fn proof_request<'a>(
469        &self,
470        original: &GitMutationRequest,
471        action: ProofAction,
472        proof_id: [u8; 16],
473        alias_digest: [u8; 32],
474        values: impl Iterator<Item = &'a GitSaveApproval> + Clone,
475    ) -> Result<Message, CoreError> {
476        self.proof_capability()?;
477        let (mut request, ready, original_length) = self.lookup_prefix(original, action)?;
478        let (original_transfer_id, approvals_transfer_id) = match ready {
479            message @ Message::ProofExisting { .. } => {
480                request.finish();
481                return Ok(message);
482            }
483            Message::ProofReady {
484                original_transfer_id,
485                approvals_transfer_id,
486            } => (original_transfer_id, approvals_transfer_id),
487            m => return Err(response(m)),
488        };
489        self.check_cancellation()?;
490        if !self.transport.owns_registration(&self.lease) {
491            return Err(unavailable("exact proof registration changed"));
492        }
493        let deadline = Instant::now() + Duration::from_secs(120);
494        request.header.stream_id = Some(original_transfer_id);
495        let mut upload = Upload::new(self, &mut request, Part::Original, deadline);
496        proof_codec::write_original(&mut upload, original, &self.target)?;
497        upload.finish(original_length)?;
498        match next(self, &mut request, deadline, true)?.header.message {
499            Message::Ack {} => {}
500            m => return Err(response(m)),
501        }
502        let mut count = Count(0);
503        approvals(&mut count, self, values.clone())?;
504        request.header.stream_id = Some(approvals_transfer_id.clone());
505        request.send(
506            Message::ProofApprovalsBegin {
507                transfer_id: approvals_transfer_id,
508                total_length: count.0,
509                proof_id: w::uuid(proof_id),
510                alias_digest: w::hex(&alias_digest),
511            },
512            Vec::new(),
513        )?;
514        match next(self, &mut request, deadline, true)?.header.message {
515            Message::Ack {} => {}
516            m => return Err(response(m)),
517        }
518        let mut upload = Upload::new(self, &mut request, Part::Approvals, deadline);
519        approvals(&mut upload, self, values)?;
520        upload.finish(count.0)?;
521        let result = next(self, &mut request, deadline, false)?.header.message;
522        request.finish();
523        Ok(result)
524    }
525    /// Existing receipt wins before approval iteration. Preserve the original ID on every error.
526    pub fn prepare_worktree<'a>(
527        &self,
528        original: &GitMutationRequest,
529        observation: [u8; 16],
530        alias_digest: [u8; 32],
531        approvals: impl Iterator<Item = &'a GitSaveApproval> + Clone,
532    ) -> Result<WorktreePreparation, CoreError> {
533        match self.proof_request(
534            original,
535            ProofAction::Prepare,
536            observation,
537            alias_digest,
538            approvals,
539        )? {
540            Message::ProofPrepared { plan_id } => {
541                Ok(WorktreePreparation::Prepared(w::id(&plan_id)?))
542            }
543            Message::ProofExisting { receipt } => {
544                let r = receipt.core(&self.target)?;
545                if r.id != original.operation {
546                    return Err(w::invalid());
547                }
548                Ok(WorktreePreparation::Existing(Box::new(r)))
549            }
550            m => Err(response(m)),
551        }
552    }
553    pub fn execute_worktree<'a>(
554        &self,
555        original: GitMutationRequest,
556        plan: [u8; 16],
557        alias_digest: [u8; 32],
558        approvals: impl Iterator<Item = &'a GitSaveApproval> + Clone,
559    ) -> Result<OperationReceipt, CoreError> {
560        match self.proof_request(
561            &original,
562            ProofAction::Execute,
563            plan,
564            alias_digest,
565            approvals,
566        )? {
567            Message::ProofExisting { receipt } => {
568                let r = receipt.core(&self.target)?;
569                if r.id != original.operation {
570                    return Err(w::invalid());
571                }
572                Ok(r)
573            }
574            m => Err(response(m)),
575        }
576    }
577    fn proof_cleanup(&self, message: Message) -> Result<(), CoreError> {
578        self.proof_capability()?;
579        let mut request = self.transport.start(message, None, None)?;
580        let result = request.next()?.header.message;
581        request.finish();
582        match result {
583            Message::Ack {} => Ok(()),
584            m => Err(response(m)),
585        }
586    }
587    pub fn cancel_worktree_plan(&self, plan: [u8; 16]) -> Result<(), CoreError> {
588        self.proof_cleanup(Message::ProofCancel {
589            lease_id: w::uuid(self.lease.lease.0),
590            plan_id: w::uuid(plan),
591        })
592    }
593    pub fn release_observation(&self, observation: [u8; 16]) -> Result<(), CoreError> {
594        self.proof_cleanup(Message::ProofRelease {
595            lease_id: w::uuid(self.lease.lease.0),
596            observation_id: w::uuid(observation),
597        })
598    }
599}
buntrail_engine/remote/proof.rs

buntrail_engine/remote/
proof.rs
