Skip to content

feat(mst2): persist native HTTP sessions and guard delivery #56

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Ivanbeethoven merged 1 commit into from
Oct 7, 2026
Merged

feat(mst2): persist native HTTP sessions and guard delivery #56

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 17 additions & 27 deletions src/api/router/snapshot_content.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,9 @@ use futures::stream::StreamExt;
use serde::Deserialize;
use serde_json::json;

use super::{abs_view_path, internal, mst2_error_response, request::Mst2Bytes, treeframe_response};
use super::{
abs_view_path, guarded_treeframe_response, internal, mst2_error_response, request::Mst2Bytes,
};
use crate::ceres::snapshot::{
chunks::{ChunkProjection, get_or_project},
error::{SnapshotError, SnapshotErrorCode},
Expand All @@ -22,7 +24,6 @@ use crate::ceres::snapshot::{
resolve_abs_metadata,
},
resolver::FsKind,
runtime::runtime,
view::validate_scope_relative_path,
};

Expand Down Expand Up @@ -226,9 +227,7 @@ pub(super) async fn blob_head(
headers: HeaderMap,
) -> Result<Response, Response> {
ensure(&state)?;
let ctx = runtime()
.context(&snapshot_id)
.map_err(mst2_error_response)?;
let ctx = super::request_context(&state, &snapshot_id).map_err(mst2_error_response)?;
validate_scope_relative_path(&q.path).map_err(mst2_error_response)?;
if headers.contains_key("range") {
return Err(mst2_error_response(SnapshotError::new(
Expand Down Expand Up @@ -286,9 +285,7 @@ pub(super) async fn objects(
Mst2Bytes(body): Mst2Bytes,
) -> Result<Response, Response> {
ensure(&state)?;
let ctx = runtime()
.context(&snapshot_id)
.map_err(mst2_error_response)?;
let ctx = super::request_context(&state, &snapshot_id).map_err(mst2_error_response)?;
let req: ObjectsRequest = super::parse_json_body(&body)?;
if req.items.is_empty() || req.items.len() > OBJECT_MAX_ITEMS {
return Err(mst2_error_response(SnapshotError::new(
Expand Down Expand Up @@ -365,7 +362,7 @@ pub(super) async fn objects(
// then exactly one END. Encoding is negotiated per request.
use crate::ceres::snapshot::frame_stream::FrameStream;
let mut stream = FrameStream::new(1, encoding);
let mut out: Vec<u8> = Vec::new();
let mut out: Vec<Vec<u8>> = Vec::new();
let mut frame: Vec<([u8; 32], Vec<u8>)> = Vec::new();
let mut frame_raw = 0usize;
for (cid, data) in unique {
Expand All @@ -377,26 +374,25 @@ pub(super) async fn objects(
let bytes = stream
.object(std::mem::take(&mut frame))
.map_err(mst2_error_response)?;
out.extend_from_slice(&bytes);
out.push(bytes);
frame_raw = 0;
}
frame_raw += data.len();
frame.push((cid, data));
}
if !frame.is_empty() {
let bytes = stream.object(frame).map_err(mst2_error_response)?;
out.extend_from_slice(&bytes);
out.push(bytes);
}

let end = stream.end(
req.items.len() as u32,
seen.len() as u32,
logical_bytes,
sha256_of(&body),
);
out.extend_from_slice(&end);
out.push(end);

treeframe_response(&snapshot_id, &body, out).map_err(mst2_error_response)
guarded_treeframe_response(&state, &ctx, &snapshot_id, &body, out).map_err(mst2_error_response)
}

#[derive(Deserialize, Debug)]
Expand Down Expand Up @@ -467,9 +463,7 @@ pub(super) async fn chunk_map(
Query(q): Query<ChunkMapQuery>,
) -> Result<Response, Response> {
ensure(&state)?;
let ctx = runtime()
.context(&snapshot_id)
.map_err(mst2_error_response)?;
let ctx = super::request_context(&state, &snapshot_id).map_err(mst2_error_response)?;
validate_scope_relative_path(&q.path).map_err(mst2_error_response)?;
let handler = state
.api_handler(std::path::Path::new("/"))
Expand Down Expand Up @@ -513,9 +507,7 @@ pub(super) async fn chunk_map_pages(
Query(q): Query<ChunkMapQuery>,
) -> Result<Response, Response> {
ensure(&state)?;
let ctx = runtime()
.context(&snapshot_id)
.map_err(mst2_error_response)?;
let ctx = super::request_context(&state, &snapshot_id).map_err(mst2_error_response)?;
validate_scope_relative_path(&q.path).map_err(mst2_error_response)?;
// Canonical v3 uses `map_id` + `page_index`; retain parsing of the old
// names only while the legacy client is still present in this checkout.
Expand Down Expand Up @@ -622,9 +614,7 @@ pub(super) async fn chunks(
Mst2Bytes(body): Mst2Bytes,
) -> Result<Response, Response> {
ensure(&state)?;
let ctx = runtime()
.context(&snapshot_id)
.map_err(mst2_error_response)?;
let ctx = super::request_context(&state, &snapshot_id).map_err(mst2_error_response)?;
let req: ChunksRequest = super::parse_json_body(&body)?;
if req.items.is_empty() || req.items.len() > CHUNKS_MAX_ITEMS {
return Err(mst2_error_response(SnapshotError::new(
Expand Down Expand Up @@ -707,7 +697,7 @@ pub(super) async fn chunks(

use crate::ceres::snapshot::frame_stream::FrameStream;
let mut stream = FrameStream::new(1, encoding);
let mut out: Vec<u8> = Vec::new();
let mut out: Vec<Vec<u8>> = Vec::new();
for p in planned.iter() {
// Re-verified slice (digest + length) from the staged projection.
let bytes = p
Expand All @@ -722,17 +712,17 @@ pub(super) async fn chunks(
bytes.to_vec(),
)
.map_err(mst2_error_response)?;
out.extend_from_slice(&frame);
out.push(frame);
}
let end = stream.end(
req.items.len() as u32,
planned.len() as u32,
logical_bytes,
sha256_of(&body),
);
out.extend_from_slice(&end);
out.push(end);

treeframe_response(&snapshot_id, &body, out).map_err(mst2_error_response)
guarded_treeframe_response(&state, &ctx, &snapshot_id, &body, out).map_err(mst2_error_response)
}

/// Strict decimal-string parse for unsigned counts (spec 04 §1: no leading
Expand Down
69 changes: 58 additions & 11 deletions src/api/router/snapshot_content_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,10 @@ use crate::{
object_storage::{MegaObjectStorageWrapper, build_object_storage},
push_queue_storage::{ClaimOutcome, EnqueueOutcome},
},
tests::{test_redis_manager, test_storage_with_config, with_test_vault},
tests::{
TestSchemaGuard, test_db_config, test_redis_manager, test_storage_with_config,
with_test_vault,
},
},
orbit_api::{
error::OrbitResult,
Expand All @@ -76,6 +79,9 @@ use crate::{

const TOKEN: &str = "mst2-fixed-content-test";

#[path = "snapshot_session_tests.rs"]
mod durable_sessions;

#[derive(Default)]
struct ReadCounts {
whole: AtomicUsize,
Expand Down Expand Up @@ -214,6 +220,7 @@ struct Fixture {
digest: [u8; 32],
counts: Arc<ReadCounts>,
_temp: tempfile::TempDir,
_schema: Option<TestSchemaGuard>,
}

fn tree(items: Vec<TreeItem>) -> Tree {
Expand Down Expand Up @@ -301,6 +308,14 @@ async fn publish_native_push(

impl Fixture {
async fn new() -> Self {
Self::new_with_pg_config(false).await
}

async fn new_with_pg_config(rebuildable: bool) -> Self {
Self::new_with_pg_config_and_directories(rebuildable, 0).await
}

async fn new_with_pg_config_and_directories(rebuildable: bool, directory_count: usize) -> Self {
let temp = tempfile::tempdir().unwrap();
let mut config = isolated_config(temp.path().join("config"));
config.monorepo.push_policy = PushPolicy::Trunk;
Expand All @@ -310,7 +325,25 @@ impl Fixture {
config.mst2.auth_token = Some(TOKEN.to_string());
let backend = build_object_storage(&config.object_storage).await.unwrap();
let counts = Arc::new(ReadCounts::default());
let mut storage = test_storage_with_config(temp.path(), config).await;
let (mut storage, schema) = if rebuildable {
let (database, schema) = test_db_config(temp.path()).await;
config.database = database;
let connection = crate::jupiter::storage::init::database_connection(&config.database)
.await
.unwrap();
(
crate::jupiter::storage::Storage::new_with_connection(
Arc::new(config),
Arc::new(connection),
backend.clone(),
)
.await
.unwrap(),
Some(schema),
)
} else {
(test_storage_with_config(temp.path(), config).await, None)
};
storage.git_service = GitService {
obj_storage: MegaObjectStorageWrapper::new(Arc::new(CountingStorage {
inner: backend,
Expand Down Expand Up @@ -344,15 +377,32 @@ impl Fixture {
let empty_oid = ObjectHash::from_hex_for_kind(HashKind::Sha1, &empty_oid).unwrap();
let empty_dir = tree(vec![]);
let nested = tree(vec![item(TreeItemMode::Blob, blob_oid, "file")]);
let project = tree(vec![
let mut project_items = vec![
item(TreeItemMode::Blob, blob_oid, "alias"),
item(TreeItemMode::Tree, empty_dir.id, "directory"),
item(TreeItemMode::Blob, empty_oid, "empty"),
item(TreeItemMode::BlobExecutable, blob_oid, "executable"),
item(TreeItemMode::Blob, blob_oid, "file"),
item(TreeItemMode::Link, link_oid, "link"),
item(TreeItemMode::Tree, nested.id, "nested"),
]);
];
let mut extra_trees = Vec::new();
for index in 0..directory_count {
// Distinct names make distinct canonical pages even though all
// directories share the same already-verified immutable blob.
let child = tree(vec![item(
TreeItemMode::Blob,
blob_oid,
&format!("file-{index:03}"),
)]);
project_items.push(item(
TreeItemMode::Tree,
child.id,
&format!("wide-{index:03}"),
));
extra_trees.push(child);
}
let project = tree(project_items);
let old_tip = Commit::from_tree_id_with_kind(
HashKind::Sha1,
project.id,
Expand All @@ -379,13 +429,9 @@ impl Fixture {
)
.unwrap();
let mono = storage.mono_storage();
mono.save_mega_trees(
vec![empty_dir, nested, project, root.clone()],
commit.id,
None,
)
.await
.unwrap();
let mut trees = vec![empty_dir, nested, project, root.clone()];
trees.extend(extra_trees);
mono.save_mega_trees(trees, commit.id, None).await.unwrap();
mono.save_mega_commits(vec![commit.clone(), old_tip.clone(), new_tip.clone()], None)
.await
.unwrap();
Expand Down Expand Up @@ -474,6 +520,7 @@ impl Fixture {
raw,
counts,
_temp: temp,
_schema: schema,
}
}

Expand Down
Loading
Loading