Skip to content

Commit 74ea969

Browse files
committed
feat(traces): derive sealed publication facts from verified raw closure
Consume the controlled PACK with its original deadline and cancellation, verify raw Git object types and references, derive ordered commit and checkpoint facts, and bind membership to the reachable closure. Seal internal publication bridges while preserving the existing atomic transaction, live authority, expected-old-tip and idempotency checks. MTR-98 independent patch v0.42.69. HTTP/SSH wiring and per-commit checkpoint history protection remain later cards; no schema change. A3: exact publication6/6 and seven complete modules (64 unique cases), fmt/strict Clippy and current visibility/DAG checks PASS. Historical A1/A2 FAIL and STOP retained. Local Claude ER05/H0 PASS session f771163d-c57e-431a-a211-6fb5446fba6f. Review SHA256: bfde2ac7486a652f1b63dba260b1bcccc19c5811065700b584e5e5bee9cd8c1b Final C: exact fmt/clippy/full-test/build/build-tests all exit0, 20 targets / 3081 passed / 0 failed / 3 existing ignored / 0 filtered. C terminal SHA256: e8f78c6d0b92457b0f039becce73f160e60fe42d4a55a3e12d073d8ea5bcc935 Frozen final-input manifest SHA256: e8aa9391f4d72185a5a191ccd6d1d912ff995d98e429e604c61d734e555b2774 Tag/Release and applicable main/tag Docker D are still pending. Signed-off-by: ELi Ma <eli@patch.sh>
1 parent 526be0c commit 74ea969

15 files changed

Lines changed: 2954 additions & 48 deletions

File tree

‎Cargo.lock‎

Lines changed: 1 addition & 1 deletion
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎Cargo.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
[package]
22
name = "mega2"
3-
version = "0.42.68"
3+
version = "0.42.69"
44
edition = "2024"
55
license = "MIT OR Apache-2.0"
66
description = "Monorepo engine and Git hosting server for AI agents: trunk-based storage, Git Smart HTTP, Git LFS, OCI registry, and Agent Session Capture"

‎docs/plan/README.md‎

Lines changed: 21 additions & 1 deletion
Large diffs are not rendered by default.

‎docs/plan/plan-20261015.md‎

Lines changed: 101 additions & 24 deletions
Large diffs are not rendered by default.

‎docs/plan/plan-long.md‎

Lines changed: 22 additions & 2 deletions
Large diffs are not rendered by default.

‎docs/plan/plan-status.md‎

Lines changed: 22 additions & 2 deletions
Large diffs are not rendered by default.

‎docs/refactoring/protocol.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -975,3 +975,5 @@ acknowledgments 之前运行;后者只在确定发包前运行。v0 中 have
975975
pkt-line。错误字段与完整应答表见 [错误文档](../errors.md#megaerrorviewunavailable--viewpackrejected视图协议错误plan-20261002)
976976
和 [history-projection 设计 §6.1「错误契约」](history-projection.md#61-url-与路由决策)。视图 URL 的错误产生方和
977977
黑盒覆盖由后续入口与打包卡登记,本节不重复场景覆盖表行。
978+
979+
**2026-10-24(MTR-98 本地A/B已核,自身审查与发布待)**:内部发布事实由受控 PACK 与本 stream 已发布 raw 闭包推导,原始字节、取消、schema 归档及可见性边界见 [Traces 唯一协议源](./traces.md)。封闭事实桥尚未接入 HTTP/SSH;逐提交 checkpoint 历史资格仍待 MTR-101,不能据此放行 wire 发布。

‎docs/refactoring/traces.md‎

Lines changed: 36 additions & 0 deletions
Large diffs are not rendered by default.

‎src/ceres/traces/admission.rs‎

Lines changed: 146 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ use crate::{
1515
#[derive(Debug, Clone)]
1616
pub struct IngestAdmission {
1717
started: Instant,
18+
continuation_cancel: std::sync::Arc<std::sync::atomic::AtomicBool>,
1819
#[cfg(test)]
1920
worker_done: Option<std::sync::Arc<std::sync::atomic::AtomicBool>>,
2021
#[cfg(test)]
@@ -38,6 +39,7 @@ impl IngestAdmission {
3839
pub fn start() -> Self {
3940
Self {
4041
started: Instant::now(),
42+
continuation_cancel: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
4143
#[cfg(test)]
4244
worker_done: None,
4345
#[cfg(test)]
@@ -77,6 +79,8 @@ impl IngestAdmission {
7779
/// Cancel this worker and drop the counters it was holding.
7880
pub fn cancel(&mut self) {
7981
self.cancelled = true;
82+
self.continuation_cancel
83+
.store(true, std::sync::atomic::Ordering::Release);
8084
self.compressed_bytes = 0;
8185
self.decoded_bytes = 0;
8286
self.objects = 0;
@@ -94,6 +98,9 @@ impl IngestAdmission {
9498
pub fn admit_ingest(admission: &IngestAdmission, max_object_bytes: u64) -> Result<(), MegaError> {
9599
let elapsed = Duration::from_secs(admission.elapsed_secs).max(admission.started.elapsed());
96100
if admission.cancelled
101+
|| admission
102+
.continuation_cancel
103+
.load(std::sync::atomic::Ordering::Acquire)
97104
|| elapsed > Duration::from_secs(TracesProtocolLimits::INGEST_TIMEOUT_SECONDS)
98105
{
99106
return Err(MegaError::Traces(TracesError::QueryTimeout));
@@ -677,11 +684,73 @@ mod tests {
677684
);
678685
}
679686

687+
#[tokio::test]
688+
async fn wire_ingest_lifetime_continuation_preserves_deadline_and_external_cancel() {
689+
use std::sync::{
690+
Arc,
691+
atomic::{AtomicBool, Ordering},
692+
};
693+
694+
use crate::ceres::traces::publication::tests::Fixture;
695+
let mut fixture = Fixture::new(git_internal::hash::HashKind::Sha1);
696+
fixture.tree(&[]);
697+
let pack = fixture.pack(None);
698+
let mut admission = IngestAdmission::start();
699+
admission.started = Instant::now() - Duration::from_secs(1);
700+
let decoded = decode_wire_pack_with_budget(
701+
&mut admission,
702+
fixture.kind,
703+
pack.as_slice(),
704+
PackBudget::for_test(4096, 4096, 4096, 10, 10, 2),
705+
)
706+
.await
707+
.unwrap();
708+
assert!(decoded.continuation().deadline <= admission.started + Duration::from_secs(2));
709+
tokio::time::sleep(Duration::from_millis(1100)).await;
710+
assert!(matches!(
711+
decoded.continuation().check(),
712+
Err(MegaError::Traces(TracesError::QueryTimeout))
713+
));
714+
let flag = Arc::new(AtomicBool::new(false));
715+
let decoded = decode_wire_pack(
716+
&mut IngestAdmission::start(),
717+
fixture.kind,
718+
pack.as_slice(),
719+
Some(flag.clone()),
720+
)
721+
.await
722+
.unwrap();
723+
let cancel = async {
724+
tokio::task::yield_now().await;
725+
flag.store(true, Ordering::Release);
726+
};
727+
let (result, ()) = tokio::join!(
728+
decoded
729+
.continuation()
730+
.run(std::future::pending::<Result<(), MegaError>>()),
731+
cancel
732+
);
733+
assert!(matches!(
734+
result,
735+
Err(MegaError::Traces(TracesError::QueryTimeout))
736+
));
737+
assert!(matches!(
738+
decoded.continuation().checkpoint().await,
739+
Err(MegaError::Traces(TracesError::QueryTimeout))
740+
));
741+
}
742+
680743
#[test]
681744
fn bounded_pack_debug_does_not_expose_payload() {
682745
let payload = b"not-for-logs".repeat(256);
683746
let rendered_payload = format!("{payload:?}");
684747
let decoded = BoundedDecodedPack {
748+
hash_kind: git_internal::hash::HashKind::Sha1,
749+
continuation: IngestContinuation {
750+
deadline: Instant::now() + Duration::from_secs(300),
751+
cancelled: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
752+
external_cancel: None,
753+
},
685754
decoded_bytes: payload.len() as u64,
686755
objects: 1,
687756
entries: vec![(
@@ -788,6 +857,7 @@ pub(super) struct PackControl {
788857
deadline: Instant,
789858
stop: std::sync::Arc<std::sync::atomic::AtomicBool>,
790859
cancel: Option<std::sync::Arc<std::sync::atomic::AtomicBool>>,
860+
request_cancel: std::sync::Arc<std::sync::atomic::AtomicBool>,
791861
}
792862

793863
impl PackControl {
@@ -805,12 +875,14 @@ impl PackControl {
805875
deadline,
806876
stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
807877
cancel: None,
878+
request_cancel: admission.continuation_cancel.clone(),
808879
})
809880
}
810881

811882
pub(super) fn check(&self) -> Result<(), MegaError> {
812883
use std::sync::atomic::Ordering;
813884
if self.stop.load(Ordering::Acquire)
885+
|| self.request_cancel.load(Ordering::Acquire)
814886
|| self
815887
.cancel
816888
.as_ref()
@@ -880,9 +952,62 @@ impl Drop for StopWorker<'_> {
880952

881953
/// A result can only be constructed by the controlled decode implementation.
882954
pub struct BoundedDecodedPack {
883-
entries: Vec<(git_internal::internal::object::types::ObjectType, Vec<u8>)>,
955+
entries: DecodedEntries,
884956
decoded_bytes: u64,
885957
objects: u64,
958+
hash_kind: git_internal::hash::HashKind,
959+
continuation: IngestContinuation,
960+
}
961+
962+
pub(super) type DecodedEntries = Vec<(git_internal::internal::object::types::ObjectType, Vec<u8>)>;
963+
964+
/// Request lifetime after the decoder has joined. Worker shutdown is separate.
965+
#[derive(Clone)]
966+
pub(crate) struct IngestContinuation {
967+
deadline: Instant,
968+
cancelled: std::sync::Arc<std::sync::atomic::AtomicBool>,
969+
external_cancel: Option<std::sync::Arc<std::sync::atomic::AtomicBool>>,
970+
}
971+
972+
impl IngestContinuation {
973+
pub(crate) fn check(&self) -> Result<(), MegaError> {
974+
use std::sync::atomic::Ordering;
975+
if Instant::now() > self.deadline
976+
|| self.cancelled.load(Ordering::Acquire)
977+
|| self
978+
.external_cancel
979+
.as_ref()
980+
.is_some_and(|flag| flag.load(Ordering::Acquire))
981+
{
982+
return Err(timeout());
983+
}
984+
Ok(())
985+
}
986+
987+
pub(crate) async fn checkpoint(&self) -> Result<(), MegaError> {
988+
self.check()?;
989+
tokio::task::yield_now().await;
990+
self.check()
991+
}
992+
993+
pub(crate) async fn run<T>(
994+
&self,
995+
future: impl std::future::Future<Output = Result<T, MegaError>>,
996+
) -> Result<T, MegaError> {
997+
self.check()?;
998+
tokio::pin!(future);
999+
let mut timer = tokio::time::interval(Duration::from_millis(5));
1000+
loop {
1001+
tokio::select! {
1002+
biased;
1003+
_ = timer.tick() => self.check()?,
1004+
result = &mut future => {
1005+
self.check()?;
1006+
return result;
1007+
}
1008+
}
1009+
}
1010+
}
8861011
}
8871012

8881013
impl std::fmt::Debug for BoundedDecodedPack {
@@ -906,6 +1031,20 @@ impl BoundedDecodedPack {
9061031
pub fn object_count(&self) -> u64 {
9071032
self.objects
9081033
}
1034+
1035+
pub(crate) fn continuation(&self) -> &IngestContinuation {
1036+
&self.continuation
1037+
}
1038+
1039+
pub(super) fn into_parts(
1040+
self,
1041+
) -> (
1042+
git_internal::hash::HashKind,
1043+
DecodedEntries,
1044+
IngestContinuation,
1045+
) {
1046+
(self.hash_kind, self.entries, self.continuation)
1047+
}
9091048
}
9101049

9111050
#[derive(Default)]
@@ -1116,6 +1255,12 @@ fn decode_worker(
11161255
entries: sink.inner.entries,
11171256
decoded_bytes: sink.decoded_bytes,
11181257
objects: sink.objects,
1258+
hash_kind: kind,
1259+
continuation: IngestContinuation {
1260+
deadline: control.deadline,
1261+
cancelled: sink.admission.continuation_cancel.clone(),
1262+
external_cancel: control.cancel.clone(),
1263+
},
11191264
})
11201265
}
11211266
Err(PackError::Budget("deadline")) => Err(timeout()),

0 commit comments

Comments
 (0)