diff --git a/Cargo.lock b/Cargo.lock index 271a53e4..fac68f1a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1100,7 +1100,7 @@ dependencies = [ [[package]] name = "graft-sdk" -version = "0.3.0" +version = "0.3.1" dependencies = [ "blake3", "graft", @@ -1115,7 +1115,7 @@ dependencies = [ [[package]] name = "graft-sdk-node" -version = "0.3.0" +version = "0.3.1" dependencies = [ "graft-sdk", "napi", diff --git a/crates/graft-sdk-node/Cargo.toml b/crates/graft-sdk-node/Cargo.toml index 8399c247..04c0194f 100644 --- a/crates/graft-sdk-node/Cargo.toml +++ b/crates/graft-sdk-node/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "graft-sdk-node" -version = "0.3.0" +version = "0.3.1" edition = "2024" authors.workspace = true license.workspace = true @@ -18,7 +18,7 @@ crate-type = ["cdylib"] workspace = true [dependencies] -graft-sdk = { path = "../graft-sdk", version = "0.3.0", features = ["bundled-sqlite"] } +graft-sdk = { path = "../graft-sdk", version = "0.3.1", features = ["bundled-sqlite"] } napi = { workspace = true } napi-derive = { workspace = true } serde_json = "1.0" diff --git a/crates/graft-sdk/Cargo.toml b/crates/graft-sdk/Cargo.toml index 043a2acf..33315c73 100644 --- a/crates/graft-sdk/Cargo.toml +++ b/crates/graft-sdk/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "graft-sdk" -version = "0.3.0" +version = "0.3.1" edition = "2024" authors.workspace = true license.workspace = true diff --git a/crates/graft-sqlite/src/pragma/repo_core.rs b/crates/graft-sqlite/src/pragma/repo_core.rs index 40f7cb90..081c87ae 100644 --- a/crates/graft-sqlite/src/pragma/repo_core.rs +++ b/crates/graft-sqlite/src/pragma/repo_core.rs @@ -89,14 +89,15 @@ pub(super) fn run_repo_clone( .remote_default_branch("origin")? .unwrap_or(repo.default_branch()?), }; - let fetch = repo.fetch("origin", &branch)?; + let clone_fetch = repo.fetch_for_clone("origin", &branch)?; + let fetch = &clone_fetch.fetch; repo.branch_create(&branch, Some(&format!("refs/remotes/origin/{branch}")))?; repo.set_branch_upstream(&branch, "origin", &branch)?; let plan = repo.plan_switch_branch(&branch)?; file.attach_repo(repo.clone())?; attached = true; let runtime = file.runtime().clone(); - let remote = Arc::new(repo.remote_store("origin")?); + let remote = clone_fetch.remote(); let plan = prepare_repo_checkout_plan(&runtime, &plan, Some(remote.clone()))?; let previous_files = BTreeMap::new(); let previous_artifacts = BTreeMap::new(); @@ -119,8 +120,8 @@ pub(super) fn run_repo_clone( remote: remote_info, current_head, current_branch, - branch: fetch.branch, - head: fetch.head, + branch: fetch.branch.clone(), + head: fetch.head.clone(), commits: fetch.commits, graft_dir: graft_dir.clone(), paths, diff --git a/crates/graft/src/remote.rs b/crates/graft/src/remote.rs index 81a4ccef..bef6d5fe 100644 --- a/crates/graft/src/remote.rs +++ b/crates/graft/src/remote.rs @@ -1,7 +1,10 @@ use std::{ collections::{HashMap, HashSet}, - env, fmt, future, + env, fmt, fs, future, + io::Write, ops::Range, + path::{Component, Path}, + pin::Pin, sync::{ Arc, atomic::{AtomicU64, Ordering}, @@ -39,6 +42,9 @@ const RECEIVE_PACK_HEADER_PACK_BYTES: &str = "x-graft-pack-bytes"; const RECEIVE_PACK_HEADER_INDEX_BYTES: &str = "x-graft-index-bytes"; const RECEIVE_PACK_HEADER_REPLACEMENT_HEX: &str = "x-graft-ref-replacement-hex"; const RECEIVE_BUNDLE_HEADER_MANIFEST_BYTES: &str = "x-graft-bundle-manifest-bytes"; +const MAX_UPLOAD_BUNDLE_MANIFEST_BYTES: usize = 64 * 1024; +const MAX_UPLOAD_BUNDLE_OBJECTS: usize = 65_536; +const MAX_UPLOAD_BUNDLE_PATH_BYTES: usize = 768; static HTTP_REQUEST_SEQUENCE: AtomicU64 = AtomicU64::new(1); enum RemotePath<'a> { @@ -73,6 +79,9 @@ pub enum RemoteErr { #[error("HTTP remote transport error: {0}")] HttpTransport(reqwest::Error), + #[error("upload-bundle filesystem error: {0}")] + UploadBundleIo(#[from] std::io::Error), + #[error("HTTP remote returned {status} for `{path}`: {message}")] HttpStatus { status: u16, @@ -470,6 +479,26 @@ struct ReceiveBundleManifestObject<'a> { allow_existing: bool, } +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct UploadBundleManifest { + version: u8, + reference: UploadBundleReference, + objects: usize, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct UploadBundleReference { + path: String, + value_hex: String, +} + +pub(crate) enum UploadBundleOutcome { + Downloaded, + Unsupported, +} + impl RemoteObjectPack { pub(crate) fn new(id: String, pack: Bytes, index: Bytes) -> Self { Self { @@ -745,6 +774,17 @@ impl Remote { } } + pub(crate) async fn download_upload_bundle( + &self, + ref_path: &str, + root: &Path, + ) -> Result { + match &self.backend { + RemoteBackend::Http(remote) => remote.download_upload_bundle(ref_path, root).await, + RemoteBackend::ObjectStore(_) => Ok(UploadBundleOutcome::Unsupported), + } + } + #[tracing::instrument(level = "trace", err(level = "debug"), skip(self, bytes))] pub async fn put_raw(&self, path: &str, bytes: impl Into) -> Result<()> { match &self.backend { @@ -1303,6 +1343,63 @@ impl HttpRemote { } } + async fn download_upload_bundle( + &self, + ref_path: &str, + root: &Path, + ) -> Result { + let response = self + .send( + self.request( + reqwest::Method::POST, + self.raw_url("upload-bundle", ref_path), + ) + .header(reqwest::header::CONTENT_LENGTH, 0) + .timeout(Duration::from_secs(30 * 60)), + "upload_bundle", + Some(0), + ) + .await?; + Self::check_protocol(&response, ref_path)?; + if matches!(response.status().as_u16(), 404 | 405) { + Self::drain_response(response).await?; + return Ok(UploadBundleOutcome::Unsupported); + } + let response = Self::check_response(response, ref_path).await?; + let manifest_bytes = upload_bundle_manifest_length(response.headers(), ref_path)?; + let mut body = HttpDownloadBody::new(response.bytes_stream()); + let manifest = body.read_exact(manifest_bytes, ref_path).await?; + let manifest = decode_upload_bundle_manifest(&manifest, ref_path)?; + validate_upload_bundle_manifest(&manifest, ref_path)?; + fs::create_dir_all(root)?; + + let mut previous_path: Option = None; + for _ in 0..manifest.objects { + let header = body.read_exact(12, ref_path).await?; + let path_bytes = u32::from_be_bytes(header[..4].try_into().unwrap()) as usize; + let object_bytes = u64::from_be_bytes(header[4..12].try_into().unwrap()); + if !(1..=MAX_UPLOAD_BUNDLE_PATH_BYTES).contains(&path_bytes) { + return Err(upload_bundle_error(ref_path, "invalid object path length")); + } + let path = body.read_exact(path_bytes, ref_path).await?; + let path = decode_upload_bundle_path(&path, previous_path.as_deref(), ref_path)?; + let destination = root.join(&path); + if let Some(parent) = destination.parent() { + fs::create_dir_all(parent)?; + } + let mut file = fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(destination)?; + body.copy_exact(object_bytes, &mut file, ref_path).await?; + file.flush()?; + previous_path = Some(path); + } + body.require_end(ref_path).await?; + write_upload_bundle_ref(root, &manifest.reference, ref_path)?; + Ok(UploadBundleOutcome::Downloaded) + } + async fn put_raw(&self, path: &str, bytes: Bytes) -> Result<()> { let request_bytes = bytes.len() as u64; let response = self @@ -1580,6 +1677,220 @@ impl HttpRemote { } } +type DownloadStream = + Pin> + Send + 'static>>; + +struct HttpDownloadBody { + stream: DownloadStream, + buffered: Bytes, +} + +impl HttpDownloadBody { + fn new( + stream: impl Stream> + Send + 'static, + ) -> Self { + Self { + stream: Box::pin(stream), + buffered: Bytes::new(), + } + } + + async fn read_exact(&mut self, length: usize, path: &str) -> Result> { + let mut output = Vec::with_capacity(length); + while output.len() < length { + let chunk = self + .next_chunk() + .await? + .ok_or_else(|| upload_bundle_error(path, "upload-bundle response is truncated"))?; + let needed = length - output.len(); + if chunk.len() <= needed { + output.extend_from_slice(&chunk); + } else { + output.extend_from_slice(&chunk[..needed]); + self.buffered = chunk.slice(needed..); + } + } + Ok(output) + } + + async fn copy_exact( + &mut self, + mut remaining: u64, + file: &mut fs::File, + path: &str, + ) -> Result<()> { + while remaining != 0 { + let chunk = self + .next_chunk() + .await? + .ok_or_else(|| upload_bundle_error(path, "upload-bundle object is truncated"))?; + let take = usize::try_from(remaining.min(chunk.len() as u64)).unwrap(); + file.write_all(&chunk[..take])?; + remaining -= take as u64; + if take != chunk.len() { + self.buffered = chunk.slice(take..); + } + } + Ok(()) + } + + async fn require_end(&mut self, path: &str) -> Result<()> { + if self.next_chunk().await?.is_some() { + return Err(upload_bundle_error( + path, + "upload-bundle response has trailing bytes", + )); + } + Ok(()) + } + + async fn next_chunk(&mut self) -> Result> { + if !self.buffered.is_empty() { + return Ok(Some(std::mem::take(&mut self.buffered))); + } + loop { + match self.stream.next().await { + Some(Ok(bytes)) if bytes.is_empty() => {} + Some(Ok(bytes)) => return Ok(Some(bytes)), + Some(Err(err)) => return Err(RemoteErr::HttpTransport(err)), + None => return Ok(None), + } + } + } +} + +fn upload_bundle_manifest_length( + headers: &reqwest::header::HeaderMap, + path: &str, +) -> Result { + let value = headers + .get(RECEIVE_BUNDLE_HEADER_MANIFEST_BYTES) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.parse::().ok()) + .filter(|length| (1..=MAX_UPLOAD_BUNDLE_MANIFEST_BYTES).contains(length)); + value.ok_or_else(|| upload_bundle_error(path, "invalid upload-bundle manifest length")) +} + +fn decode_upload_bundle_manifest(bytes: &[u8], path: &str) -> Result { + serde_json::from_slice(bytes) + .map_err(|err| upload_bundle_error(path, format!("invalid upload-bundle manifest: {err}"))) +} + +fn validate_upload_bundle_manifest(manifest: &UploadBundleManifest, path: &str) -> Result<()> { + if manifest.version != 1 { + return Err(upload_bundle_error( + path, + "unsupported upload-bundle manifest version", + )); + } + if manifest.reference.path != path { + return Err(upload_bundle_error( + path, + "upload-bundle reference path does not match request", + )); + } + if manifest.objects > MAX_UPLOAD_BUNDLE_OBJECTS { + return Err(upload_bundle_error( + path, + "upload-bundle contains too many objects", + )); + } + validate_upload_bundle_path(path, true) + .map_err(|message| upload_bundle_error(path, message))?; + let _ = decode_lower_hex(&manifest.reference.value_hex, path)?; + Ok(()) +} + +fn decode_upload_bundle_path( + bytes: &[u8], + previous: Option<&str>, + request: &str, +) -> Result { + let path = std::str::from_utf8(bytes) + .map_err(|_| upload_bundle_error(request, "upload-bundle object path is not UTF-8"))?; + validate_upload_bundle_path(path, false) + .map_err(|message| upload_bundle_error(request, message))?; + if previous.is_some_and(|previous| previous.as_bytes() >= path.as_bytes()) { + return Err(upload_bundle_error( + request, + "upload-bundle object paths are not ordered", + )); + } + Ok(path.to_string()) +} + +fn validate_upload_bundle_path( + path: &str, + transactional: bool, +) -> std::result::Result<(), &'static str> { + if path.is_empty() || path.len() > MAX_UPLOAD_BUNDLE_PATH_BYTES || path.contains('\\') { + return Err("upload-bundle contains an invalid path"); + } + if path + .split('/') + .any(|segment| segment.is_empty() || matches!(segment, "." | "..")) + || path.chars().any(char::is_control) + || Path::new(path) + .components() + .any(|component| !matches!(component, Component::Normal(_))) + { + return Err("upload-bundle contains an unsafe path"); + } + let is_transactional = path == "HEAD" || path.starts_with("refs/"); + if is_transactional != transactional || path == "locks" || path.starts_with("locks/") { + return Err("upload-bundle path has the wrong storage class"); + } + Ok(()) +} + +fn write_upload_bundle_ref( + root: &Path, + reference: &UploadBundleReference, + request: &str, +) -> Result<()> { + let bytes = decode_lower_hex(&reference.value_hex, request)?; + let destination = root.join(&reference.path); + if let Some(parent) = destination.parent() { + fs::create_dir_all(parent)?; + } + let mut file = fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(destination)?; + file.write_all(&bytes)?; + file.flush()?; + Ok(()) +} + +fn decode_lower_hex(value: &str, path: &str) -> Result> { + if value.len() > 32 * 1024 + || !value.len().is_multiple_of(2) + || !value + .bytes() + .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase()) + { + return Err(upload_bundle_error( + path, + "upload-bundle reference is not lowercase hexadecimal", + )); + } + (0..value.len()) + .step_by(2) + .map(|offset| u8::from_str_radix(&value[offset..offset + 2], 16)) + .collect::, _>>() + .map_err(|_| { + upload_bundle_error(path, "upload-bundle reference is not lowercase hexadecimal") + }) +} + +fn upload_bundle_error(path: &str, message: impl Into) -> RemoteErr { + RemoteErr::HttpStatus { + status: 502, + path: path.to_string(), + message: message.into(), + } +} + fn safe_server_timings(headers: &reqwest::header::HeaderMap) -> Vec<(&'static str, f64)> { let mut timings = Vec::new(); for value in headers.get_all("server-timing") { @@ -1773,6 +2084,42 @@ mod tests { (format!("http://{address}/org/repo"), task) } + async fn serve_upload_bundle( + manifest: serde_json::Value, + objects: &[(&str, &[u8])], + ) -> (String, tokio::task::JoinHandle>) { + let manifest = serde_json::to_vec(&manifest).unwrap(); + let body = encode_test_upload_bundle(&manifest, objects); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let task = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let request = read_http_request(&mut stream).await; + let headers = format!( + "HTTP/1.1 200 OK\r\nGraft-Protocol: 1\r\n{RECEIVE_BUNDLE_HEADER_MANIFEST_BYTES}: {}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + manifest.len(), + body.len(), + ); + stream.write_all(headers.as_bytes()).await.unwrap(); + for chunk in body.chunks(3) { + stream.write_all(chunk).await.unwrap(); + } + request + }); + (format!("http://{address}/org/repo"), task) + } + + fn encode_test_upload_bundle(manifest: &[u8], objects: &[(&str, &[u8])]) -> Vec { + let mut body = manifest.to_vec(); + for (path, bytes) in objects { + body.extend_from_slice(&(path.len() as u32).to_be_bytes()); + body.extend_from_slice(&(bytes.len() as u64).to_be_bytes()); + body.extend_from_slice(path.as_bytes()); + body.extend_from_slice(bytes); + } + body + } + async fn read_http_request(stream: &mut tokio::net::TcpStream) -> Vec { let mut request = Vec::new(); let mut request_bytes = None; @@ -1815,6 +2162,100 @@ mod tests { &request[header_end + 4..] } + #[tokio::test] + async fn upload_bundle_downloads_one_stream_into_a_local_remote() { + let manifest = serde_json::json!({ + "version": 1, + "reference": { + "path": "refs/heads/main", + "value_hex": hex_encode(b"commit-one\n"), + }, + "objects": 2, + }); + let (url, request) = serve_upload_bundle( + manifest, + &[ + ("objects/pack/example.idx", b"index"), + ("objects/pack/example.pack", b"pack"), + ], + ) + .await; + let remote = RemoteConfig::Http { url, token_env: None }.build().unwrap(); + let destination = tempfile::tempdir().unwrap(); + + assert!(matches!( + remote + .download_upload_bundle("refs/heads/main", destination.path()) + .await + .unwrap(), + UploadBundleOutcome::Downloaded + )); + assert_eq!( + fs::read(destination.path().join("refs/heads/main")).unwrap(), + b"commit-one\n" + ); + assert_eq!( + fs::read(destination.path().join("objects/pack/example.idx")).unwrap(), + b"index" + ); + assert_eq!( + fs::read(destination.path().join("objects/pack/example.pack")).unwrap(), + b"pack" + ); + assert!( + request + .await + .unwrap() + .starts_with(b"POST /org/repo/upload-bundle/refs/heads/main ") + ); + } + + #[tokio::test] + async fn upload_bundle_rejects_paths_outside_the_destination() { + let manifest = serde_json::json!({ + "version": 1, + "reference": { + "path": "refs/heads/main", + "value_hex": hex_encode(b"commit-one\n"), + }, + "objects": 1, + }); + let (url, request) = serve_upload_bundle(manifest, &[("../escaped", b"bad")]).await; + let remote = RemoteConfig::Http { url, token_env: None }.build().unwrap(); + let parent = tempfile::tempdir().unwrap(); + let destination = parent.path().join("bundle"); + + assert!(matches!( + remote + .download_upload_bundle("refs/heads/main", &destination) + .await, + Err(RemoteErr::HttpStatus { status: 502, .. }) + )); + assert!(!parent.path().join("escaped").exists()); + request.await.unwrap(); + } + + #[tokio::test] + async fn upload_bundle_falls_back_when_the_remote_does_not_support_it() { + let (url, request) = serve_http_response("404 Not Found", &["1"]).await; + let remote = RemoteConfig::Http { url, token_env: None }.build().unwrap(); + let destination = tempfile::tempdir().unwrap(); + + assert!(matches!( + remote + .download_upload_bundle("refs/heads/main", destination.path()) + .await + .unwrap(), + UploadBundleOutcome::Unsupported + )); + assert!( + request + .await + .unwrap() + .starts_with("POST /org/repo/upload-bundle/refs/heads/main ") + ); + } + #[tokio::test] async fn receive_pack_publishes_pack_index_and_ref_in_one_request() { let (url, requests) = serve_http_exchanges(&["204 No Content"]).await; diff --git a/crates/graft/src/repo.rs b/crates/graft/src/repo.rs index a2249d85..71363e81 100644 --- a/crates/graft/src/repo.rs +++ b/crates/graft/src/repo.rs @@ -67,7 +67,7 @@ use crate::{ LogId, VolumeId, byte_unit::ByteUnit, commit_hash::CommitHash, lsn::LSN, lsn::LSNRangeExt, page_count::PageCount, }, - remote::{RemoteConfig, RemoteCredentials, RemoteErr}, + remote::{Remote, RemoteConfig, RemoteCredentials, RemoteErr, UploadBundleOutcome}, snapshot::Snapshot, }; @@ -421,6 +421,18 @@ pub struct FetchOutcome { pub commits: usize, } +pub struct CloneFetch { + pub fetch: FetchOutcome, + remote: Arc, + _bundle: Option, +} + +impl CloneFetch { + pub fn remote(&self) -> Arc { + self.remote.clone() + } +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct FetchAllOutcome { pub remote: String, diff --git a/crates/graft/src/repo/remote_objects.rs b/crates/graft/src/repo/remote_objects.rs index 7abaa1b9..61292257 100644 --- a/crates/graft/src/repo/remote_objects.rs +++ b/crates/graft/src/repo/remote_objects.rs @@ -105,7 +105,7 @@ impl Repository { remote: &crate::remote::Remote, id: &object::ObjectId, pack_cache: &mut RemoteObjectPackCache, - ) -> Result { + ) -> Result> { let hit = pack_cache.indexes(remote)?.iter().find_map(|index| { index .objects @@ -114,10 +114,7 @@ impl Repository { .map(|entry| (index.pack.clone(), entry.offset, entry.len)) }); let Some((pack, offset, len)) = hit else { - return Err(RepoErr::InvalidRemoteObject { - path: object::LooseObjectStore::relative_path(id), - message: "missing object".to_string(), - }); + return Ok(None); }; let end = offset .checked_add(len) @@ -140,7 +137,7 @@ impl Repository { message: format!("pack entry for object {id} extends past pack length"), }); } - Ok(pack_bytes.slice(offset..end)) + Ok(Some(pack_bytes.slice(offset..end))) } pub(super) fn fetch_commit_chain( @@ -450,9 +447,14 @@ impl Repository { pack_cache: &mut RemoteObjectPackCache, ) -> Result { let path = object::LooseObjectStore::relative_path(id); - let bytes = match block_on_remote(remote.get_raw(&path))? { + let bytes = match self.fetch_packed_object_bytes(remote, id, pack_cache)? { Some(bytes) => bytes, - None => self.fetch_packed_object_bytes(remote, id, pack_cache)?, + None => block_on_remote(remote.get_raw(&path))?.ok_or_else(|| { + RepoErr::InvalidRemoteObject { + path: path.clone(), + message: "missing object".to_string(), + } + })?, }; Ok(self.object_store().write_raw_validated(id, &bytes)?) } diff --git a/crates/graft/src/repo/sync.rs b/crates/graft/src/repo/sync.rs index 03aec926..96c9bedb 100644 --- a/crates/graft/src/repo/sync.rs +++ b/crates/graft/src/repo/sync.rs @@ -5,6 +5,50 @@ impl Repository { validate_remote_name(remote)?; validate_ref_name(branch)?; let remote_store = self.remote_store(remote)?; + self.fetch_with_store(remote, branch, &remote_store) + } + + pub fn fetch_for_clone(&self, remote: &str, branch: &str) -> Result { + validate_remote_name(remote)?; + validate_ref_name(branch)?; + let network_remote = self.remote_store(remote)?; + let head_path = format!("refs/heads/{branch}"); + let bundle = tempfile::tempdir()?; + let outcome = + block_on_remote(network_remote.download_upload_bundle(&head_path, bundle.path()))?; + match outcome { + UploadBundleOutcome::Downloaded => { + let root = bundle.path().to_string_lossy().into_owned(); + let remote_store = RemoteConfig::Fs { root }.build()?; + let fetch = self.fetch_with_store(remote, branch, &remote_store)?; + Ok(CloneFetch { + fetch, + remote: Arc::new(remote_store), + _bundle: Some(bundle), + }) + } + UploadBundleOutcome::Unsupported => { + let fetch = self.fetch_with_store(remote, branch, &network_remote)?; + // Do not carry a pooled HTTP/1 connection from large pack downloads into + // checkout. Some proxies leave that connection readable but unusable for + // the first storage-commit request. A new pool preserves the legacy fallback. + self.remote_credentials.reset_http_clients(); + let checkout_remote = self.remote_store(remote)?; + Ok(CloneFetch { + fetch, + remote: Arc::new(checkout_remote), + _bundle: None, + }) + } + } + } + + fn fetch_with_store( + &self, + remote: &str, + branch: &str, + remote_store: &Remote, + ) -> Result { let head_path = format!("refs/heads/{branch}"); let Some(head) = block_on_remote(remote_store.get_raw(&head_path))? else { return Err(RepoErr::RemoteBranchNotFound { @@ -13,7 +57,7 @@ impl Repository { }); }; let head = parse_remote_ref(&head_path, head)?; - let commits = self.fetch_commit_chain(&remote_store, &head)?; + let commits = self.fetch_commit_chain(remote_store, &head)?; self.set_remote_tracking_ref(remote, branch, &head)?; Ok(FetchOutcome { remote: remote.to_string(), diff --git a/docs/src/content/docs/docs/reference/remote-protocol.mdx b/docs/src/content/docs/docs/reference/remote-protocol.mdx index 1fc0d6c9..e926abe4 100644 --- a/docs/src/content/docs/docs/reference/remote-protocol.mdx +++ b/docs/src/content/docs/docs/reference/remote-protocol.mdx @@ -90,6 +90,7 @@ issuance, tenancy, and ACL storage are service concerns outside this protocol. "range", "list", "put-if-absent", + "upload-bundle", "receive-pack", "receive-bundle", "cas", @@ -138,6 +139,7 @@ All paths below are relative to `{base}`. | `PUT /raw/` | Replace transactional metadata. | `204 No Content` | | `DELETE /raw/` | Delete transactional metadata. | `204 No Content` | | `PUT /raw-if-not-exists/` | Create an object only when absent. | `204 No Content` | +| `POST /upload-bundle/` | Stream a ref snapshot and its immutable storage. | `200 OK` | | `POST /receive-pack/` | Publish one object pack and atomically update a ref. | `204 No Content` | | `POST /receive-bundle/` | Publish immutable objects, a pack, and a ref. | `204 No Content` | | `POST /cas/` | Atomically compare and replace metadata. | `204 No Content` | @@ -158,6 +160,48 @@ objects but cannot expose an incomplete commit. Deleting an already absent key through `DELETE /raw` SHOULD succeed so the operation remains idempotent. +## Upload Bundle + +`upload-bundle` is an optional version 1 capability for Git-style clone +transport. A client selects a branch and sends one authenticated request: + +```http +POST /upload-bundle/refs/heads/main +Content-Length: 0 +``` + +The service reads the requested ref, lists the repository's immutable keys, +then reads the ref again. If it changed, the service returns `409`; the client +may retry. A stable snapshot returns +`Content-Type: application/vnd.graft.upload-bundle` and the +`x-graft-bundle-manifest-bytes` header. The response starts with that many +bytes of UTF-8 JSON: + +```json +{ + "version": 1, + "reference": { + "path": "refs/heads/main", + "value_hex": "" + }, + "objects": 3 +} +``` + +Exactly `objects` binary frames follow. Each frame is a 4-byte unsigned path +length, an 8-byte unsigned object length, the UTF-8 path, then the object body. +Integers use network byte order. Paths MUST be strictly ordered, unique, +immutable repository keys. The body MUST end immediately after the final +frame. The service streams each immutable backend object without buffering the +whole bundle. + +Version 1 bundles all immutable keys because the protocol service treats their +contents as opaque. The client validates every frame, creates a temporary local +remote, and resolves the selected commit graph and checkout from that local +snapshot. This gives clone one bulk data response; future protocol versions may +add reachability negotiation. Clients MUST fall back to the version 1 raw/list +operations when `upload-bundle` returns `404` or `405`. + ## Receive Pack `receive-pack` is an optional version 1 capability that collapses pack upload, diff --git a/docs/src/content/docs/zh/docs/reference/remote-protocol.mdx b/docs/src/content/docs/zh/docs/reference/remote-protocol.mdx index 690d73f9..33c53878 100644 --- a/docs/src/content/docs/zh/docs/reference/remote-protocol.mdx +++ b/docs/src/content/docs/zh/docs/reference/remote-protocol.mdx @@ -76,6 +76,7 @@ ACL 存储不属于本协议。 "range", "list", "put-if-absent", + "upload-bundle", "receive-pack", "receive-bundle", "cas", @@ -120,6 +121,7 @@ encoding 必须被拒绝。服务可以公布 key/body 限制，并在超限时 | `PUT /raw/` | 替换事务元数据。 | `204 No Content` | | `DELETE /raw/` | 删除事务元数据。 | `204 No Content` | | `PUT /raw-if-not-exists/` | 仅当对象不存在时创建。 | `204 No Content` | +| `POST /upload-bundle/` | 流式返回 ref 快照及其不可变存储。 | `200 OK` | | `POST /receive-pack/` | 发布一个 object pack 并原子更新 ref。 | `204 No Content` | | `POST /receive-bundle/` | 发布不可变对象、pack 并原子更新 ref。 | `204 No Content` | | `POST /cas/` | 原子 compare-and-swap。 | `204 No Content` | @@ -136,6 +138,43 @@ encoding 必须被拒绝。服务可以公布 key/body 限制，并在超限时 对已不存在 key 的 `DELETE /raw` 应该仍成功，以保持幂等。 +## Upload Bundle + +`upload-bundle` 是 version 1 的可选 capability，用于 Git 风格的 clone 传输。client +选定 branch 后只发送一个已认证请求： + +```http +POST /upload-bundle/refs/heads/main +Content-Length: 0 +``` + +服务先读取请求的 ref，列举 repository 的不可变 key，再读取一次 ref。如果 ref +已经变化则返回 `409`，client 可以重试。快照稳定时返回 +`Content-Type: application/vnd.graft.upload-bundle` 和 +`x-graft-bundle-manifest-bytes` header。响应开头是该 header 指定长度的 UTF-8 JSON： + +```json +{ + "version": 1, + "reference": { + "path": "refs/heads/main", + "value_hex": "" + }, + "objects": 3 +} +``` + +之后必须紧跟 `objects` 个二进制 frame。每个 frame 依次包含 4 字节无符号 path +长度、8 字节无符号对象长度、UTF-8 path 和对象 body；整数使用网络字节序。path +必须严格有序、唯一，并且属于不可变 repository key。最后一个 frame 后不得有多余 +字节。服务直接转发每个 backend 对象流，无需把整个 bundle 缓存在内存中。 + +version 1 会打包全部不可变 key，因为协议服务把对象内容视为 opaque。client 校验 +所有 frame，建立临时本地 remote，再从该本地快照解析选定 commit graph 并完成 +checkout。因此 clone 的数据主体只有一个 bulk response；后续协议版本可以增加 +reachability negotiation。`upload-bundle` 返回 `404` 或 `405` 时，client 必须回退 +到 version 1 的 raw/list 操作。 + ## Receive Pack `receive-pack` 是 version 1 的可选 capability，把 pack 上传、index 上传和最终 ref diff --git a/packages/graft-remote-cloudflare/package.json b/packages/graft-remote-cloudflare/package.json index 3d6c7b5e..ef05db99 100644 --- a/packages/graft-remote-cloudflare/package.json +++ b/packages/graft-remote-cloudflare/package.json @@ -1,6 +1,6 @@ { "name": "@eidos.space/graft-remote-cloudflare", - "version": "0.1.0", + "version": "0.2.0", "description": "Cloudflare storage and authentication adapters for @eidos.space/graft-remote", "type": "module", "license": "MIT", diff --git a/packages/graft-remote-hono/package.json b/packages/graft-remote-hono/package.json index 121c7a0f..3ea65512 100644 --- a/packages/graft-remote-hono/package.json +++ b/packages/graft-remote-hono/package.json @@ -1,6 +1,6 @@ { "name": "@eidos.space/graft-remote-hono", - "version": "0.1.0", + "version": "0.2.0", "description": "Hono adapter for @eidos.space/graft-remote", "type": "module", "license": "MIT", diff --git a/packages/graft-remote/README.md b/packages/graft-remote/README.md index 7fd29fb2..3886c5f2 100644 --- a/packages/graft-remote/README.md +++ b/packages/graft-remote/README.md @@ -116,6 +116,8 @@ Backend guarantees: - `list` returns at most `query.limit` paths, sorted by UTF-8 byte order, after `query.after`, and restricted to `query.prefix`. - Immutable request bodies remain streams when the adapter storage supports it. +- `upload-bundle` lists immutable keys and streams each `get` body into one + clone response without buffering the repository in memory. - `options.contentLength` is the exact length of a framed immutable body, such as each object in a `receive-pack` or `receive-bundle` request. - Repository instances are isolated by `repository.id`. diff --git a/packages/graft-remote/package.json b/packages/graft-remote/package.json index 013c110e..4b78a4a0 100644 --- a/packages/graft-remote/package.json +++ b/packages/graft-remote/package.json @@ -1,6 +1,6 @@ { "name": "@eidos.space/graft-remote", - "version": "0.1.0", + "version": "0.2.0", "description": "Framework-neutral types and protocol engine for Graft remote services", "type": "module", "license": "MIT", diff --git a/packages/graft-remote/src/handler.ts b/packages/graft-remote/src/handler.ts index b2c80be6..04396039 100644 --- a/packages/graft-remote/src/handler.ts +++ b/packages/graft-remote/src/handler.ts @@ -1,10 +1,14 @@ import { GRAFT_REMOTE_CAPABILITIES, + MAX_LIST_LIMIT, MAX_METADATA_BYTES, + MAX_UPLOAD_BUNDLE_OBJECTS, PROTOCOL_HEADER, PROTOCOL_VERSION, + RECEIVE_BUNDLE_HEADER_MANIFEST_BYTES, GraftProtocolError, bytewiseCompare, + bytesEqual, emptyResponse, encodeListCursor, errorResponse, @@ -41,12 +45,14 @@ import type { const OPERATIONS = new Set([ "raw", "raw-if-not-exists", + "upload-bundle", "receive-pack", "receive-bundle", "cas", "cad", "list", ]); +const UPLOAD_BUNDLE_PREFETCH_OBJECTS = 8; export function createGraftRemoteHandler( options: GraftRemoteOptions, @@ -144,6 +150,8 @@ async function handleRequest( return raw(input.request, backend, path); case "raw-if-not-exists": return putIfAbsent(input.request, backend, path); + case "upload-bundle": + return uploadBundle(backend, path); case "receive-pack": return receivePack(input.request, backend, path); case "receive-bundle": @@ -207,6 +215,9 @@ function validateMethodAndAction( case "raw-if-not-exists": requireMethod(method, "PUT"); return "write"; + case "upload-bundle": + requireMethod(method, "POST"); + return "read"; case "receive-pack": case "receive-bundle": requireMethod(method, "POST"); @@ -310,6 +321,268 @@ async function putIfAbsent( return emptyResponse(); } +async function uploadBundle( + backend: GraftRepositoryBackend, + refPath: string, +): Promise { + requireTransactionalPath(refPath); + const reference = await readMetadataObject(backend, refPath); + if (reference === null) throw objectNotFound(); + const paths = await listImmutablePaths(backend); + const confirmed = await readMetadataObject(backend, refPath); + if (confirmed === null || !bytesEqual(reference, confirmed)) { + throw new GraftProtocolError( + 409, + "snapshot_changed", + "Reference changed while preparing the upload bundle", + ); + } + + const manifest = new TextEncoder().encode( + JSON.stringify({ + version: 1, + reference: { path: refPath, value_hex: encodeLowerHex(reference) }, + objects: paths.length, + }), + ); + const headers = protocolHeaders({ + "Content-Type": "application/vnd.graft.upload-bundle", + [RECEIVE_BUNDLE_HEADER_MANIFEST_BYTES]: manifest.byteLength.toString(), + }); + return new Response(new UploadBundleBody(backend, manifest, paths).stream, { headers }); +} + +async function listImmutablePaths(backend: GraftRepositoryBackend): Promise { + const paths: string[] = []; + let after: string | undefined; + for (;;) { + const result = await backend.list({ prefix: "", after, limit: MAX_LIST_LIMIT }); + if (result.paths.length > MAX_LIST_LIMIT) { + throw backendContractError("List backend returned more paths than requested"); + } + for (const path of result.paths) { + validateObjectPath(path); + if (after !== undefined && bytewiseCompare(path, after) <= 0) { + throw backendContractError("List backend returned unsorted paths"); + } + after = path; + if (isImmutablePath(path)) paths.push(path); + if (paths.length > MAX_UPLOAD_BUNDLE_OBJECTS) { + throw new GraftProtocolError(413, "upload_bundle_too_large", "Too many bundled objects"); + } + } + if (!result.hasMore) return paths; + if (result.paths.length === 0) { + throw backendContractError("List backend cannot advance the upload-bundle cursor"); + } + } +} + +async function readMetadataObject( + backend: GraftRepositoryBackend, + path: string, +): Promise | null> { + const object = await backend.get(path); + if (object === null) return null; + validateMetadata(object); + if (object.size > MAX_METADATA_BYTES) { + throw backendContractError("Transactional backend object exceeds the metadata limit"); + } + const bytes = await readObjectBody(object, MAX_METADATA_BYTES); + if (bytes.byteLength !== object.size) { + throw backendContractError("Backend object body does not match its declared size"); + } + return bytes; +} + +async function readObjectBody( + object: GraftObject, + limit: number, +): Promise> { + const reader = objectBodyStream(object).getReader(); + const chunks: Uint8Array[] = []; + let total = 0; + try { + for (;;) { + const result = await reader.read(); + if (result.done) break; + total += result.value.byteLength; + if (total > limit) throw backendContractError("Backend object body exceeds its size limit"); + chunks.push(result.value); + } + } finally { + reader.releaseLock(); + } + const bytes = new Uint8Array(new ArrayBuffer(total)); + let offset = 0; + for (const chunk of chunks) { + bytes.set(chunk, offset); + offset += chunk.byteLength; + } + return bytes; +} + +function encodeLowerHex(bytes: Uint8Array): string { + let value = ""; + for (const byte of bytes) value += byte.toString(16).padStart(2, "0"); + return value; +} + +function objectBodyStream(object: GraftObject): ReadableStream { + if (object.body instanceof ReadableStream) return object.body; + const bytes = object.body instanceof Uint8Array ? object.body : new Uint8Array(object.body); + return new ReadableStream({ + start(controller) { + controller.enqueue(bytes); + controller.close(); + }, + }); +} + +class UploadBundleBody { + readonly stream: ReadableStream; + readonly #backend: GraftRepositoryBackend; + readonly #paths: string[]; + readonly #pending: Uint8Array[]; + readonly #prefetched = new Map>(); + #index = 0; + #nextPrefetch = 0; + #reader: ReadableStreamDefaultReader | undefined; + #remaining = 0; + + constructor(backend: GraftRepositoryBackend, manifest: Uint8Array, paths: string[]) { + this.#backend = backend; + this.#paths = paths; + this.#pending = [manifest]; + this.stream = new ReadableStream({ + pull: async (controller) => { + try { + await this.pull(controller); + } catch (error) { + await this.cancel(error); + throw error; + } + }, + cancel: async (reason) => await this.cancel(reason), + }); + } + + private async pull(controller: ReadableStreamDefaultController): Promise { + for (;;) { + const pending = this.#pending.shift(); + if (pending !== undefined) { + controller.enqueue(pending); + return; + } + if (this.#reader !== undefined) { + if (await this.pullObject(controller)) return; + continue; + } + if (this.#index === this.#paths.length) { + controller.close(); + return; + } + await this.openObject(); + } + } + + private async openObject(): Promise { + this.fillPrefetch(); + const path = this.#paths[this.#index]!; + const prefetched = await this.#prefetched.get(this.#index)!; + this.#prefetched.delete(this.#index); + if (!prefetched.ok) throw prefetched.error; + const object = prefetched.object; + if (object === null) throw backendContractError("Bundled immutable object disappeared"); + validateMetadata(object); + this.#remaining = object.size; + this.#reader = objectBodyStream(object).getReader(); + this.#pending.push(uploadBundleFrameHeader(path, object.size)); + } + + private fillPrefetch(): void { + const end = Math.min( + this.#paths.length, + this.#index + UPLOAD_BUNDLE_PREFETCH_OBJECTS, + ); + while (this.#nextPrefetch < end) { + const index = this.#nextPrefetch; + const path = this.#paths[index]!; + const object = Promise.resolve(this.#backend.get(path)).then( + (value): PrefetchedObject => ({ ok: true, object: value }), + (error: unknown): PrefetchedObject => ({ ok: false, error }), + ); + this.#prefetched.set(index, object); + this.#nextPrefetch += 1; + } + } + + private async pullObject( + controller: ReadableStreamDefaultController, + ): Promise { + for (;;) { + const result = await this.#reader!.read(); + if (result.done) { + this.#reader!.releaseLock(); + this.#reader = undefined; + if (this.#remaining !== 0) throw backendContractError("Bundled object body is truncated"); + this.#index += 1; + return false; + } + if (result.value.byteLength === 0) continue; + if (result.value.byteLength > this.#remaining) { + throw backendContractError("Bundled object body exceeds its declared size"); + } + this.#remaining -= result.value.byteLength; + controller.enqueue(result.value); + return true; + } + } + + private async cancel(reason: unknown): Promise { + const reader = this.#reader; + this.#reader = undefined; + if (reader !== undefined) { + try { + await reader.cancel(reason); + } finally { + reader.releaseLock(); + } + } + + const prefetched = [...this.#prefetched.values()]; + this.#prefetched.clear(); + this.#nextPrefetch = this.#paths.length; + const objects = await Promise.all(prefetched); + await Promise.all( + objects.map(async (result) => { + if (!result.ok || result.object === null) return; + const body = result.object.body; + if (!(body instanceof ReadableStream) || body.locked) return; + try { + await body.cancel(reason); + } catch { + // A backend may have already canceled the prefetched body. + } + }), + ); + } +} + +type PrefetchedObject = + | { ok: true; object: GraftObject | null } + | { ok: false; error: unknown }; + +function uploadBundleFrameHeader(path: string, size: number): Uint8Array { + const pathBytes = new TextEncoder().encode(path); + const header = new Uint8Array(new ArrayBuffer(12 + pathBytes.byteLength)); + const view = new DataView(header.buffer); + view.setUint32(0, pathBytes.byteLength); + view.setBigUint64(4, BigInt(size)); + header.set(pathBytes, 12); + return header; +} + async function receivePack( request: Request, backend: GraftRepositoryBackend, diff --git a/packages/graft-remote/src/index.ts b/packages/graft-remote/src/index.ts index 15f3237a..f4f74889 100644 --- a/packages/graft-remote/src/index.ts +++ b/packages/graft-remote/src/index.ts @@ -4,6 +4,7 @@ export { GRAFT_REMOTE_CAPABILITIES, MAX_LIST_LIMIT, MAX_METADATA_BYTES, + MAX_UPLOAD_BUNDLE_OBJECTS, PROTOCOL_HEADER, PROTOCOL_VERSION, RECEIVE_PACK_HEADER_INDEX_BYTES, @@ -11,6 +12,7 @@ export { RECEIVE_PACK_HEADER_PACK_ID, RECEIVE_PACK_HEADER_REPLACEMENT_HEX, RECEIVE_PACK_ID_BYTES, + RECEIVE_BUNDLE_HEADER_MANIFEST_BYTES, GraftProtocolError, bytesEqual, bytewiseCompare, diff --git a/packages/graft-remote/src/protocol.ts b/packages/graft-remote/src/protocol.ts index 722b824d..0cb1136a 100644 --- a/packages/graft-remote/src/protocol.ts +++ b/packages/graft-remote/src/protocol.ts @@ -10,6 +10,7 @@ export const GRAFT_REMOTE_CAPABILITIES = [ "list", "list-cursor", "put-if-absent", + "upload-bundle", "receive-pack", "receive-bundle", "cas", @@ -24,6 +25,7 @@ export const RECEIVE_PACK_HEADER_INDEX_BYTES = "x-graft-index-bytes"; export const RECEIVE_PACK_HEADER_REPLACEMENT_HEX = "x-graft-ref-replacement-hex"; export const RECEIVE_BUNDLE_HEADER_MANIFEST_BYTES = "x-graft-bundle-manifest-bytes"; export const MAX_RECEIVE_BUNDLE_OBJECTS = 256; +export const MAX_UPLOAD_BUNDLE_OBJECTS = 65_536; const REPOSITORY_SEGMENT = /^[A-Za-z0-9](?:[A-Za-z0-9._-]{0,62}[A-Za-z0-9])?$/; const encoder = new TextEncoder(); const decoder = new TextDecoder("utf-8", { fatal: true }); diff --git a/packages/graft-remote/src/types.ts b/packages/graft-remote/src/types.ts index 8624f947..1a65802a 100644 --- a/packages/graft-remote/src/types.ts +++ b/packages/graft-remote/src/types.ts @@ -69,6 +69,7 @@ export type GraftRemoteOperation = | "descriptor" | "raw" | "raw-if-not-exists" + | "upload-bundle" | "receive-pack" | "receive-bundle" | "cas" diff --git a/packages/graft-remote/test/remote.test.ts b/packages/graft-remote/test/remote.test.ts index 293b2a57..12e8d37d 100644 --- a/packages/graft-remote/test/remote.test.ts +++ b/packages/graft-remote/test/remote.test.ts @@ -23,7 +23,10 @@ class MemoryRepository implements GraftRepositoryBackend { return value === undefined ? null : { size: value.byteLength }; } - get(path: string, range?: GraftByteRange): GraftObject | null { + get( + path: string, + range?: GraftByteRange, + ): GraftObject | null | Promise { const value = this.objects.get(path); if (value === undefined) { return null; @@ -143,6 +146,7 @@ describe("createGraftRemoteHandler", () => { capabilities: expect.arrayContaining([ "range", "list", + "upload-bundle", "receive-pack", "receive-bundle", "cas", @@ -305,6 +309,94 @@ describe("createGraftRemoteHandler", () => { ).toBe("pack"); }); + it("streams a ref snapshot and immutable objects in one upload-bundle request", async () => { + const app = createTestApp(); + for (const [path, body] of [ + ["refs/heads/main", "commit-1\n"], + ["objects/pack/one.idx", "index"], + ["objects/pack/one.pack", "pack-data"], + ["segments/one", "segment"], + ] as const) { + const operation = path.startsWith("refs/") ? "raw" : "raw-if-not-exists"; + expect( + ( + await remoteFetch(app, `/upload/repo/${operation}/${path}`, { + method: "PUT", + body, + }) + ).status, + ).toBe(204); + } + + const response = await remoteFetch(app, "/upload/repo/upload-bundle/refs/heads/main", { + method: "POST", + }); + expect(response.status, await response.clone().text()).toBe(200); + expect(response.headers.get("content-type")).toBe("application/vnd.graft.upload-bundle"); + const bytes = new Uint8Array(await response.arrayBuffer()); + const manifestBytes = Number(response.headers.get("x-graft-bundle-manifest-bytes")); + const manifest = JSON.parse(new TextDecoder().decode(bytes.subarray(0, manifestBytes))) as { + version: number; + reference: { path: string; value_hex: string }; + objects: number; + }; + expect(manifest).toEqual({ + version: 1, + reference: { path: "refs/heads/main", value_hex: "636f6d6d69742d310a" }, + objects: 3, + }); + expect(decodeUploadBundleFrames(bytes.subarray(manifestBytes), manifest.objects)).toEqual([ + ["objects/pack/one.idx", "index"], + ["objects/pack/one.pack", "pack-data"], + ["segments/one", "segment"], + ]); + }); + + it("prefetches upload-bundle objects with a bounded concurrency window", async () => { + class DelayedRepository extends MemoryRepository { + activeImmutableGets = 0; + maximumImmutableGets = 0; + + override async get(path: string, range?: GraftByteRange): Promise { + if (!path.startsWith("refs/")) { + this.activeImmutableGets += 1; + this.maximumImmutableGets = Math.max( + this.maximumImmutableGets, + this.activeImmutableGets, + ); + await new Promise((resolve) => setTimeout(resolve, 5)); + this.activeImmutableGets -= 1; + } + return await super.get(path, range); + } + } + + const backend = new DelayedRepository(); + backend.put("refs/heads/main", new TextEncoder().encode("commit-1\n")); + for (let index = 0; index < 12; index += 1) { + backend.put( + `objects/prefetch/${index.toString().padStart(2, "0")}`, + new Uint8Array([index]), + ); + } + const app = createGraftRemoteHandler({ backend: () => backend }); + const response = await remoteFetch(app, "/prefetch/repo/upload-bundle/refs/heads/main", { + method: "POST", + }); + expect(response.status, await response.clone().text()).toBe(200); + await response.arrayBuffer(); + expect(backend.maximumImmutableGets).toBe(8); + }); + + it("returns not found when upload-bundle cannot resolve the requested ref", async () => { + const response = await remoteFetch( + createTestApp(), + "/missing/repo/upload-bundle/refs/heads/main", + { method: "POST" }, + ); + expect(response.status).toBe(404); + }); + it("does not publish a ref when a receive-pack body is truncated", async () => { const app = createTestApp(); const packId = "b".repeat(64); @@ -546,6 +638,24 @@ function joinBytes(parts: Uint8Array[]): Uint8Array { return bytes; } +function decodeUploadBundleFrames(bytes: Uint8Array, count: number): Array<[string, string]> { + const frames: Array<[string, string]> = []; + let offset = 0; + for (let index = 0; index < count; index += 1) { + const view = new DataView(bytes.buffer, bytes.byteOffset + offset); + const pathBytes = view.getUint32(0); + const bodyBytes = Number(view.getBigUint64(4)); + offset += 12; + const path = new TextDecoder().decode(bytes.subarray(offset, offset + pathBytes)); + offset += pathBytes; + const body = new TextDecoder().decode(bytes.subarray(offset, offset + bodyBytes)); + offset += bodyBytes; + frames.push([path, body]); + } + expect(offset).toBe(bytes.byteLength); + return frames; +} + function textHex(value: string): string { return [...new TextEncoder().encode(value)] .map((byte) => byte.toString(16).padStart(2, "0")) diff --git a/packages/graft-sdk/package.json b/packages/graft-sdk/package.json index 8eb32b9e..3ffd5fd5 100644 --- a/packages/graft-sdk/package.json +++ b/packages/graft-sdk/package.json @@ -1,6 +1,6 @@ { "name": "@eidos.space/graft", - "version": "0.3.0", + "version": "0.3.1", "description": "Long-lived in-process Graft repository sessions for Node.js and Electron.", "license": "MIT OR Apache-2.0", "repository": { diff --git a/packages/graft-sdk/test/repository-session.test.js b/packages/graft-sdk/test/repository-session.test.js index 77ad86a5..2d2532d4 100644 --- a/packages/graft-sdk/test/repository-session.test.js +++ b/packages/graft-sdk/test/repository-session.test.js @@ -24,7 +24,7 @@ const { } = require("..") test("exposes ABI-stable SDK metadata and materialization contract", () => { - assert.equal(sdkVersion(), "0.3.0") + assert.equal(sdkVersion(), "0.3.1") for (const operation of [ "restore", "restorePaths", diff --git a/scripts/remote-release.test.mjs b/scripts/remote-release.test.mjs index 769dc0b0..097b5d83 100644 --- a/scripts/remote-release.test.mjs +++ b/scripts/remote-release.test.mjs @@ -60,7 +60,7 @@ test("validates the checked-in Remote package release contract", async () => { ); metadataByName.set(metadata.name, metadata); } - validatePackageMetadata(metadataByName, "0.1.0"); + validatePackageMetadata(metadataByName, "0.2.0"); }); test("rejects a dependency that could escape the release version", () => { diff --git a/services/graft-remote-cloudflare/test/worker.test.ts b/services/graft-remote-cloudflare/test/worker.test.ts index 68be3a66..b25c8ee4 100644 --- a/services/graft-remote-cloudflare/test/worker.test.ts +++ b/services/graft-remote-cloudflare/test/worker.test.ts @@ -22,6 +22,27 @@ async function responseText(response: Response): Promise { return new TextDecoder().decode(await response.arrayBuffer()); } +function decodeUploadBundle(bytes: Uint8Array, manifestBytes: number): Array<[string, string]> { + const manifest = JSON.parse(new TextDecoder().decode(bytes.subarray(0, manifestBytes))) as { + objects: number; + }; + const objects: Array<[string, string]> = []; + let offset = manifestBytes; + for (let index = 0; index < manifest.objects; index += 1) { + const view = new DataView(bytes.buffer, bytes.byteOffset + offset); + const pathBytes = view.getUint32(0); + const bodyBytes = Number(view.getBigUint64(4)); + offset += 12; + const path = new TextDecoder().decode(bytes.subarray(offset, offset + pathBytes)); + offset += pathBytes; + const body = new TextDecoder().decode(bytes.subarray(offset, offset + bodyBytes)); + offset += bodyBytes; + objects.push([path, body]); + } + expect(offset).toBe(bytes.byteLength); + return objects; +} + describe("graft remote protocol", () => { it("requires authentication and protocol negotiation", async () => { const unauthorized = await SELF.fetch(`${ORIGIN}/auth/repo`, { @@ -93,6 +114,33 @@ describe("graft remote protocol", () => { expect(head.headers.get("content-length")).toBe("6"); }); + it("streams an upload bundle across transactional storage and R2", async () => { + await remoteFetch("/upload/repo/raw/refs/heads/main", { + method: "PUT", + body: "commit-one\n", + }); + await remoteFetch("/upload/repo/raw-if-not-exists/objects/pack/one.idx", { + method: "PUT", + body: "index", + }); + await remoteFetch("/upload/repo/raw-if-not-exists/objects/pack/one.pack", { + method: "PUT", + body: "pack", + }); + + const response = await remoteFetch("/upload/repo/upload-bundle/refs/heads/main", { + method: "POST", + }); + expect(response.status).toBe(200); + const manifestBytes = Number(response.headers.get("x-graft-bundle-manifest-bytes")); + expect(manifestBytes).toBeGreaterThan(0); + const bytes = new Uint8Array(await response.arrayBuffer()); + expect(decodeUploadBundle(bytes, manifestBytes)).toEqual([ + ["objects/pack/one.idx", "index"], + ["objects/pack/one.pack", "pack"], + ]); + }); + it("performs atomic ref compare-and-swap and compare-and-delete", async () => { const ref = "/cas/repo/cas/refs/heads/main"; const created = await remoteFetch(ref, {