You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by ps...@apache.org on 2026/08/18 02:44:20 UTC
(opendal-ofs) branch rewrite/harness created (now 62e7280)
This is an automated email from the ASF dual-hosted git repository.
PsiACE pushed a change to branch rewrite/harness
in repository https://gitbox.apache.org/repos/asf/opendal-ofs.git
at 62e7280 test(harness): drive random-write scale through FileChangeSet
This branch includes the following new commits:
new 05d96c5 feat(harness): add black-box Managed Sync behavior suite
new 824b71c docs: document Managed Sync behavior and scale cargo x commands
new c511fda ci: require Managed Sync behavior on pull requests
new 62e7280 test(harness): drive random-write scale through FileChangeSet
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(opendal-ofs) 01/04: feat(harness): add black-box Managed Sync behavior suite
Posted by ps...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
PsiACE pushed a commit to branch rewrite/harness
in repository https://gitbox.apache.org/repos/asf/opendal-ofs.git
commit 05d96c5b26b31aeb70c891da6efcb4acba34d033
Author: PsiACE <ps...@apache.org>
AuthorDate: Tue Aug 18 02:13:38 2026 +0800
feat(harness): add black-box Managed Sync behavior suite
One binary, no --profile. Cases create RFC volume names and assert
tree, status, conflict, recovery, GC, FastCDC/Zstd, Branch, and chaos.
---
Cargo.lock | 5 +
Cargo.toml | 11 +-
.../managed-sync/bub.Dockerfile | 36 +-
fixtures/managed-sync/compose.yaml | 76 ++
xtask/Cargo.toml | 5 +
xtask/src/main.rs | 129 +++
xtask/src/managed_sync.rs | 30 +
xtask/src/managed_sync/behavior.rs | 105 +++
xtask/src/managed_sync/behavior/lifecycle.rs | 679 ++++++++++++++++
xtask/src/managed_sync/behavior/reconciliation.rs | 229 ++++++
xtask/src/managed_sync/behavior/recovery.rs | 144 ++++
xtask/src/managed_sync/bub.rs | 497 ++++++++++++
xtask/src/managed_sync/cli.rs | 275 +++++++
xtask/src/managed_sync/evaluation.rs | 279 +++++++
xtask/src/managed_sync/fixture.rs | 890 +++++++++++++++++++++
xtask/src/managed_sync/scale.rs | 675 ++++++++++++++++
xtask/src/managed_sync/scale/random_write.rs | 325 ++++++++
17 files changed, 4369 insertions(+), 21 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 4e90ab8..8e007b2 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2560,7 +2560,12 @@ checksum = "3ad82d2a33cdc9674dc7465672f271e096168fcdbe0f799d9e6db8c5892679dc"
name = "x"
version = "0.0.0"
dependencies = [
+ "blake3",
"clap",
+ "serde",
+ "serde_json",
+ "tempfile",
+ "walkdir",
"which",
]
diff --git a/Cargo.toml b/Cargo.toml
index 64eca1c..38be8ce 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -37,13 +37,18 @@ path = "src/main.rs"
[dependencies]
anyhow = "1"
clap.workspace = true
+futures.workspace = true
ofs-core = { path = "crates/core" }
-ofs-extras = { path = "crates/extras" }
ofs-ext-branch = { path = "crates/extensions/branch" }
ofs-ext-fastcdc = { path = "crates/extensions/fastcdc" }
ofs-ext-zstd = { path = "crates/extensions/zstd" }
-opendal = { version = "0.57.0", features = ["layers-chaos", "services-fs", "services-memory", "services-s3"] }
-futures.workspace = true
+ofs-extras = { path = "crates/extras" }
+opendal = { version = "0.57.0", features = [
+ "layers-chaos",
+ "services-fs",
+ "services-memory",
+ "services-s3",
+] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
diff --git a/xtask/Cargo.toml b/fixtures/managed-sync/bub.Dockerfile
similarity index 55%
copy from xtask/Cargo.toml
copy to fixtures/managed-sync/bub.Dockerfile
index f36f160..3fe8141 100644
--- a/xtask/Cargo.toml
+++ b/fixtures/managed-sync/bub.Dockerfile
@@ -15,23 +15,23 @@
# specific language governing permissions and limitations
# under the License.
-[package]
-name = "x"
-version = "0.0.0"
+FROM rust:1.91-bookworm AS ofs-builder
+ENV RUSTUP_TOOLCHAIN=1.91.1
+WORKDIR /source
+COPY Cargo.toml Cargo.lock LICENSE NOTICE README.md ./
+COPY crates ./crates
+COPY src ./src
+COPY xtask/Cargo.toml ./xtask/Cargo.toml
+COPY xtask/src/main.rs ./xtask/src/main.rs
+RUN cargo install --locked --path . --root /opt/ofs
-edition.workspace = true
-homepage.workspace = true
-license.workspace = true
-publish = false
-repository.workspace = true
-rust-version.workspace = true
+FROM python:3.13-slim-bookworm
+ARG BUB_VERSION=0.4.0
+ARG UV_VERSION=0.6.14
+RUN python -m pip install --no-cache-dir "bub==${BUB_VERSION}" "uv==${UV_VERSION}"
+COPY --from=ofs-builder /opt/ofs/bin/ofs /usr/local/bin/ofs
-[package.metadata.release]
-release = false
-
-[dependencies]
-clap = { workspace = true }
-which = { workspace = true }
-
-[lints]
-workspace = true
+ENV BUB_HOME=/sync/sessions \
+ HOME=/var/lib/bub
+WORKDIR /workspace
+CMD ["sleep", "infinity"]
diff --git a/fixtures/managed-sync/compose.yaml b/fixtures/managed-sync/compose.yaml
new file mode 100644
index 0000000..a25ba80
--- /dev/null
+++ b/fixtures/managed-sync/compose.yaml
@@ -0,0 +1,76 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+services:
+ bub-a: &bub-client
+ build:
+ context: ../..
+ dockerfile: fixtures/managed-sync/bub.Dockerfile
+ image: ${OFS_BUB_IMAGE:-opendal-ofs-managed-sync-bub:local}
+ depends_on:
+ - minio
+ network_mode: host
+ environment: &bub-environment
+ AWS_ACCESS_KEY_ID: minioadmin
+ AWS_EC2_METADATA_DISABLED: "true"
+ AWS_REGION: us-east-1
+ AWS_SECRET_ACCESS_KEY: minioadmin
+ HOME: /var/lib/bub
+ OFS_HOME: /var/lib/ofs
+ NO_PROXY: 127.0.0.1,localhost
+ no_proxy: 127.0.0.1,localhost
+
+ bub-b:
+ <<: *bub-client
+ environment:
+ <<: *bub-environment
+
+ minio:
+ image: quay.io/minio/minio:RELEASE.2024-09-22T00-33-43Z
+ command: server /data
+ environment:
+ MINIO_ROOT_PASSWORD: minioadmin
+ MINIO_ROOT_USER: minioadmin
+ NO_PROXY: minio,127.0.0.1,localhost
+ no_proxy: minio,127.0.0.1,localhost
+ ports:
+ - "127.0.0.1:${OFS_MANAGED_SYNC_MINIO_PORT:-19000}:9000"
+ volumes:
+ - minio-data:/data
+
+ minio-client:
+ image: quay.io/minio/mc:RELEASE.2024-09-16T17-43-14Z
+ entrypoint:
+ - mc
+ environment:
+ MC_HOST_local: http://minioadmin:minioadmin@minio:9000
+ MC_NO_COLOR: "1"
+ MC_QUIET: "1"
+ NO_PROXY: minio,127.0.0.1,localhost
+ no_proxy: minio,127.0.0.1,localhost
+
+ toxiproxy:
+ image: ghcr.io/shopify/toxiproxy:2.12.0
+ environment:
+ NO_PROXY: minio,toxiproxy,127.0.0.1,localhost
+ no_proxy: minio,toxiproxy,127.0.0.1,localhost
+ ports:
+ - "127.0.0.1:${OFS_MANAGED_SYNC_PROXY_PORT:-19001}:8666"
+ - "127.0.0.1:${OFS_MANAGED_SYNC_PROXY_ADMIN_PORT:-19002}:8474"
+
+volumes:
+ minio-data:
diff --git a/xtask/Cargo.toml b/xtask/Cargo.toml
index f36f160..6e66be4 100644
--- a/xtask/Cargo.toml
+++ b/xtask/Cargo.toml
@@ -30,7 +30,12 @@ rust-version.workspace = true
release = false
[dependencies]
+blake3 = { workspace = true }
clap = { workspace = true }
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+tempfile = "3"
+walkdir = "2.5"
which = { workspace = true }
[lints]
diff --git a/xtask/src/main.rs b/xtask/src/main.rs
index 94f97ea..52d4867 100644
--- a/xtask/src/main.rs
+++ b/xtask/src/main.rs
@@ -22,6 +22,8 @@ use std::process::Command as StdCommand;
use clap::Parser;
use clap::Subcommand;
+mod managed_sync;
+
fn main() {
let cmd = Command::parse();
cmd.run();
@@ -37,9 +39,25 @@ struct Command {
impl Command {
fn run(self) {
match self.sub {
+ SubCommand::Behavior {
+ target: BehaviorTarget::ManagedSync(cmd),
+ } => managed_sync::run_fixture(cmd.keep, cmd.case, cmd.evaluation),
SubCommand::Check(cmd) => cmd.run(),
+ SubCommand::E2e {
+ target: E2eTarget::ManagedSyncBub(cmd),
+ } => managed_sync::run_bub_e2e(cmd.keep),
SubCommand::Licenses(cmd) => cmd.run(),
SubCommand::Lint(cmd) => cmd.run(),
+ SubCommand::Scale {
+ target: ScaleTarget::ManagedSync(cmd),
+ } => managed_sync::run_scale(
+ cmd.scenario,
+ cmd.keep,
+ cmd.duration_seconds,
+ cmd.sync_interval_seconds,
+ cmd.file_size,
+ cmd.evaluation,
+ ),
SubCommand::Test(cmd) => cmd.run(),
}
}
@@ -47,16 +65,127 @@ impl Command {
#[derive(Subcommand)]
enum SubCommand {
+ #[clap(about = "Run user-visible behavior scenarios.")]
+ Behavior {
+ #[command(subcommand)]
+ target: BehaviorTarget,
+ },
#[clap(about = "Check all workspace targets.")]
Check(CommandCheck),
+ #[clap(about = "Run end-to-end scenarios that require external credentials.")]
+ E2e {
+ #[command(subcommand)]
+ target: E2eTarget,
+ },
#[clap(about = "Check source headers and dependency licenses.")]
Licenses(CommandLicenses),
#[clap(about = "Run source and documentation linters.")]
Lint(CommandLint),
+ #[clap(about = "Run fixed-scale acceptance scenarios.")]
+ Scale {
+ #[command(subcommand)]
+ target: ScaleTarget,
+ },
#[clap(about = "Run all workspace tests.")]
Test(CommandTest),
}
+#[derive(Subcommand)]
+enum ScaleTarget {
+ #[clap(about = "Run Managed Sync scale acceptance.")]
+ ManagedSync(CommandManagedSyncScale),
+}
+
+#[derive(Parser)]
+#[clap(name = "managed-sync")]
+struct CommandManagedSyncScale {
+ #[arg(value_enum)]
+ scenario: managed_sync::ScaleScenario,
+
+ #[arg(long, help = "Leave the fixture and generated files after completion.")]
+ keep: bool,
+
+ #[arg(
+ long,
+ default_value_t = 60,
+ help = "Random-write workload duration in seconds."
+ )]
+ duration_seconds: u64,
+
+ #[arg(
+ long,
+ default_value_t = 60,
+ help = "Random-write sync interval in seconds."
+ )]
+ sync_interval_seconds: u64,
+
+ #[arg(
+ long,
+ default_value = "1GiB",
+ value_name = "SIZE",
+ value_parser = parse_size,
+ help = "Random-write file size."
+ )]
+ file_size: u64,
+
+ #[command(flatten)]
+ evaluation: managed_sync::EvaluationOptions,
+}
+
+#[derive(Subcommand)]
+enum E2eTarget {
+ #[clap(about = "Synchronize Bub sessions and skills across isolated containers.")]
+ ManagedSyncBub(CommandManagedSyncBub),
+}
+
+#[derive(Parser)]
+#[clap(name = "managed-sync-bub")]
+struct CommandManagedSyncBub {
+ #[arg(long, help = "Leave the fixture running after the command exits.")]
+ keep: bool,
+}
+
+#[derive(Subcommand)]
+enum BehaviorTarget {
+ #[clap(about = "Run Managed Sync behavior scenarios.")]
+ ManagedSync(CommandManagedSyncBehavior),
+}
+
+#[derive(Parser)]
+#[clap(name = "managed-sync")]
+struct CommandManagedSyncBehavior {
+ #[arg(long, value_enum, value_name = "NAME")]
+ case: Option<managed_sync::BehaviorCase>,
+
+ #[arg(long, help = "Leave the fixture running after the command exits.")]
+ keep: bool,
+
+ #[command(flatten)]
+ evaluation: managed_sync::EvaluationOptions,
+}
+
+fn parse_size(value: &str) -> Result<u64, String> {
+ let value = value.trim();
+ let (digits, multiplier) = if let Some(digits) = value.strip_suffix("GiB") {
+ (digits, 1024 * 1024 * 1024)
+ } else if let Some(digits) = value.strip_suffix("MiB") {
+ (digits, 1024 * 1024)
+ } else if let Some(digits) = value.strip_suffix("KiB") {
+ (digits, 1024)
+ } else if let Some(digits) = value.strip_suffix('B') {
+ (digits, 1)
+ } else {
+ (value, 1)
+ };
+ let count = digits
+ .trim()
+ .parse::<u64>()
+ .map_err(|_| format!("invalid size {value}"))?;
+ count
+ .checked_mul(multiplier)
+ .ok_or_else(|| format!("size {value} overflows"))
+}
+
#[derive(Parser)]
#[clap(name = "check")]
struct CommandCheck {}
diff --git a/xtask/src/managed_sync.rs b/xtask/src/managed_sync.rs
new file mode 100644
index 0000000..0018d0a
--- /dev/null
+++ b/xtask/src/managed_sync.rs
@@ -0,0 +1,30 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Managed Sync acceptance harness.
+
+mod behavior;
+mod bub;
+mod cli;
+mod evaluation;
+mod fixture;
+mod scale;
+
+pub(crate) use behavior::{BehaviorCase, run as run_fixture};
+pub(crate) use bub::run as run_bub_e2e;
+pub(crate) use evaluation::EvaluationOptions;
+pub(crate) use scale::{ScaleScenario, run as run_scale};
diff --git a/xtask/src/managed_sync/behavior.rs b/xtask/src/managed_sync/behavior.rs
new file mode 100644
index 0000000..19412b3
--- /dev/null
+++ b/xtask/src/managed_sync/behavior.rs
@@ -0,0 +1,105 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! User-visible Managed Sync behavior registry.
+
+mod lifecycle;
+mod reconciliation;
+mod recovery;
+
+use std::fs;
+use std::path::Path;
+
+use super::cli::Ofs;
+use super::evaluation::EvaluationOptions;
+use super::fixture::Fixture;
+
+#[derive(Clone, Copy, Debug, clap::ValueEnum)]
+pub(crate) enum BehaviorCase {
+ Admission,
+ ContentReuse,
+ Gc,
+ Growing,
+ Extensions,
+ Branch,
+ Chaos,
+ Interruption,
+ OfflineGc,
+ Reconcile,
+ Rename,
+ Smoke,
+}
+
+pub(crate) fn run(keep: bool, case: Option<BehaviorCase>, evaluation: EvaluationOptions) {
+ let fixture = Fixture::new(keep, "behavior", evaluation);
+ let ofs = Ofs::debug(fixture.ofs_home());
+ ofs.build();
+ let fixture = fixture.start();
+ fixture.create_bucket();
+ match case {
+ Some(BehaviorCase::Admission) => lifecycle::admission(&fixture, ofs),
+ Some(BehaviorCase::ContentReuse) => lifecycle::content_reuse(&fixture, ofs),
+ Some(BehaviorCase::Gc) => recovery::gc(&fixture, ofs),
+ Some(BehaviorCase::Growing) => lifecycle::growing(&fixture, ofs),
+ Some(BehaviorCase::Extensions) => lifecycle::extensions(&fixture, ofs),
+ Some(BehaviorCase::Branch) => lifecycle::branch(&fixture, ofs),
+ Some(BehaviorCase::Chaos) => lifecycle::chaos(&fixture, ofs),
+ Some(BehaviorCase::Interruption) => recovery::interruption(&fixture, ofs),
+ Some(BehaviorCase::OfflineGc) => recovery::offline_gc(&fixture, ofs),
+ Some(BehaviorCase::Reconcile) => reconciliation::reconcile(&fixture, ofs),
+ Some(BehaviorCase::Rename) => reconciliation::rename(&fixture, ofs),
+ Some(BehaviorCase::Smoke) => lifecycle::smoke(&fixture, ofs),
+ None => {
+ lifecycle::admission(&fixture, ofs.clone());
+ lifecycle::smoke(&fixture, ofs.clone());
+ lifecycle::content_reuse(&fixture, ofs.clone());
+ lifecycle::extensions(&fixture, ofs.clone());
+ lifecycle::branch(&fixture, ofs.clone());
+ lifecycle::chaos(&fixture, ofs.clone());
+ reconciliation::reconcile(&fixture, ofs.clone());
+ reconciliation::rename(&fixture, ofs.clone());
+ recovery::offline_gc(&fixture, ofs.clone());
+ recovery::interruption(&fixture, ofs.clone());
+ lifecycle::growing(&fixture, ofs.clone());
+ recovery::gc(&fixture, ofs);
+ }
+ }
+ println!("Managed Sync behavior passed: {case:?}");
+}
+
+pub(super) fn deterministic_bytes(length: usize, seed: u8) -> Vec<u8> {
+ let mut bytes = vec![0_u8; length];
+ let mut hasher = blake3::Hasher::new();
+ hasher.update(b"ofs-managed-sync-behavior");
+ hasher.update(&[seed]);
+ hasher.finalize_xof().fill(&mut bytes);
+ bytes
+}
+
+#[cfg(unix)]
+pub(super) fn make_executable(path: &Path) {
+ use std::os::unix::fs::PermissionsExt as _;
+
+ let mut permissions = fs::metadata(path)
+ .expect("read executable metadata")
+ .permissions();
+ permissions.set_mode(permissions.mode() | 0o111);
+ fs::set_permissions(path, permissions).expect("set executable mode");
+}
+
+#[cfg(not(unix))]
+fn make_executable(_path: &Path) {}
diff --git a/xtask/src/managed_sync/behavior/lifecycle.rs b/xtask/src/managed_sync/behavior/lifecycle.rs
new file mode 100644
index 0000000..1060974
--- /dev/null
+++ b/xtask/src/managed_sync/behavior/lifecycle.rs
@@ -0,0 +1,679 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Volume admission, ordinary lifecycle, and growing-file behavior.
+
+use std::fs;
+use std::io::{Seek as _, SeekFrom, Write as _};
+
+use super::super::cli::{ManagedStatus, Ofs, output_text, require_failure, require_success};
+use super::super::fixture::{CaseRoot, Fixture, LogicalIo, tree_summary};
+use super::{deterministic_bytes, make_executable};
+
+pub(super) fn chaos(fixture: &Fixture, ofs: Ofs) {
+ let root = CaseRoot::new();
+ let a = root.path.join("a");
+ let b = root.path.join("b");
+ let state_a = root.path.join("state-a");
+ let state_b = root.path.join("state-b");
+ fs::create_dir_all(&a).expect("create chaos source");
+ fs::create_dir_all(&b).expect("create chaos destination");
+ fs::write(a.join("file"), b"chaos retry\n").expect("write chaos content");
+ let storage = fixture.storage_url("chaos");
+
+ require_success(ofs.volume_create(&storage), "create chaos volume");
+ require_success(ofs.sync(&a, &state_a, &storage), "publish chaos source");
+ let mut failed = ofs.sync(&b, &state_b, &storage);
+ failed.env("OFS_CHAOS_ERROR_RATIO", "1");
+ require_failure(failed, "inject an OpenDAL read failure");
+ require_success(ofs.sync(&b, &state_b, &storage), "retry after chaos");
+ assert_eq!(tree_summary(&a), tree_summary(&b));
+}
+
+pub(super) fn growing(fixture: &Fixture, ofs: Ofs) {
+ let root = CaseRoot::new();
+ let replica_a = root.path.join("replica-a");
+ let replica_b = root.path.join("replica-b");
+ let state_a = root.path.join("state-a");
+ let state_b = root.path.join("state-b");
+ fs::create_dir_all(&replica_a).expect("create growing replica A");
+ fs::create_dir_all(&replica_b).expect("create growing replica B");
+ let storage = fixture.storage_url("growing");
+
+ require_success(ofs.volume_create(&storage), "create growing-file volume");
+ let initial = deterministic_bytes(2 * 1024 * 1024, 17);
+ fs::write(replica_a.join("session.tape"), &initial).expect("write growing session");
+ require_success(
+ ofs.sync(&replica_a, &state_a, &storage),
+ "publish growing session",
+ );
+ require_success(
+ ofs.sync(&replica_b, &state_b, &storage),
+ "restore growing session",
+ );
+
+ let appended = deterministic_bytes(128 * 1024, 91);
+ let mut file = fs::OpenOptions::new()
+ .append(true)
+ .open(replica_a.join("session.tape"))
+ .expect("open growing session");
+ file.write_all(&appended).expect("append growing session");
+ file.sync_all().expect("persist growing session");
+ let changes = root.path.join("growing-changes.jsonl");
+ write_changes(
+ &changes,
+ "session.tape",
+ &initial,
+ &[(initial.len() as u64, appended.len() as u64)],
+ );
+ require_success(
+ ofs.sync_changes(&replica_a, &state_a, &storage, &changes),
+ "publish appended session",
+ );
+ require_success(
+ ofs.sync(&replica_b, &state_b, &storage),
+ "install appended session",
+ );
+ assert_eq!(
+ tree_summary(&replica_a),
+ tree_summary(&replica_b),
+ "an appended session converges without changing its bytes"
+ );
+ let no_op = require_success(
+ ofs.sync(&replica_b, &state_b, &storage),
+ "repeat appended session sync",
+ );
+ assert!(
+ !output_text(&no_op.stdout).contains("(published)"),
+ "an unchanged appended session is a no-op"
+ );
+
+ let mut expected = initial;
+ expected.extend_from_slice(&appended);
+ let replacement = deterministic_bytes(appended.len(), 92);
+ let mut file = fs::OpenOptions::new()
+ .write(true)
+ .open(replica_a.join("session.tape"))
+ .expect("open growing session for replacement");
+ file.seek(SeekFrom::Start((expected.len() - appended.len()) as u64))
+ .expect("seek growing session");
+ file.write_all(&replacement)
+ .expect("replace appended session range");
+ file.sync_all().expect("persist replaced session range");
+ let changes = root.path.join("replacement-changes.jsonl");
+ write_changes(
+ &changes,
+ "session.tape",
+ &expected,
+ &[(
+ (expected.len() - appended.len()) as u64,
+ appended.len() as u64,
+ )],
+ );
+ let replacement_start = expected.len() - appended.len();
+ expected[replacement_start..].copy_from_slice(&replacement);
+ require_success(
+ ofs.sync_changes(&replica_a, &state_a, &storage, &changes),
+ "publish a replacement over an appended range",
+ );
+ require_success(
+ ofs.sync(&replica_b, &state_b, &storage),
+ "install a layered replacement",
+ );
+
+ let previous = expected.clone();
+ let edge_length = 4 * 1024;
+ let head = deterministic_bytes(edge_length, 121);
+ let tail = deterministic_bytes(edge_length, 122);
+ let tail_offset = expected.len() - edge_length;
+ let mut file = fs::OpenOptions::new()
+ .write(true)
+ .open(replica_a.join("session.tape"))
+ .expect("open growing session for edge updates");
+ file.write_all(&head).expect("replace session head");
+ file.seek(SeekFrom::Start(tail_offset as u64))
+ .expect("seek session tail");
+ file.write_all(&tail).expect("replace session tail");
+ file.sync_all().expect("persist session edge updates");
+ expected[..edge_length].copy_from_slice(&head);
+ expected[tail_offset..].copy_from_slice(&tail);
+ let changes = root.path.join("edge-changes.jsonl");
+ write_changes(
+ &changes,
+ "session.tape",
+ &previous,
+ &[
+ (0, edge_length as u64),
+ (tail_offset as u64, edge_length as u64),
+ ],
+ );
+ require_success(
+ ofs.sync_changes(&replica_a, &state_a, &storage, &changes),
+ "publish disjoint head and tail changes",
+ );
+ require_success(
+ ofs.sync(&replica_b, &state_b, &storage),
+ "install disjoint head and tail changes",
+ );
+ assert_eq!(
+ fs::read(replica_b.join("session.tape")).expect("read edge-updated growing file"),
+ expected,
+ "disjoint changes preserve unchanged bytes between their ranges"
+ );
+ require_success(ofs.gc(&storage), "collect superseded growing-file data");
+ let replica_c = root.path.join("replica-c");
+ let state_c = root.path.join("state-c");
+ fs::create_dir_all(&replica_c).expect("create post-GC replica");
+ require_success(
+ ofs.sync(&replica_c, &state_c, &storage),
+ "cold-install layered growing file after collection",
+ );
+ assert_eq!(
+ fs::read(replica_c.join("session.tape")).expect("read layered growing file"),
+ expected,
+ "layered append and replacement remain readable after collection"
+ );
+}
+
+fn write_changes(path: &std::path::Path, file: &str, base: &[u8], ranges: &[(u64, u64)]) {
+ let ranges = ranges
+ .iter()
+ .map(|(offset, length)| serde_json::json!({ "offset": offset, "length": length }))
+ .collect::<Vec<_>>();
+ let record = serde_json::json!({
+ "path": file,
+ "base": {
+ "digest": blake3::hash(base).to_hex().to_string(),
+ "length": base.len(),
+ },
+ "ranges": ranges,
+ });
+ fs::write(path, format!("{record}\n")).expect("write staged changes");
+}
+
+pub(super) fn content_reuse(fixture: &Fixture, ofs: Ofs) {
+ let root = CaseRoot::new();
+ let replica_a = root.path.join("replica-a");
+ let replica_b = root.path.join("replica-b");
+ let state_a = root.path.join("state-a");
+ let state_b = root.path.join("state-b");
+ fs::create_dir_all(&replica_a).expect("create content-reuse replica A");
+ fs::create_dir_all(&replica_b).expect("create content-reuse replica B");
+ let storage = fixture.storage_url("content-reuse");
+
+ fixture.observe(
+ "content-reuse",
+ "create volume",
+ LogicalIo::default(),
+ || {
+ require_success(ofs.volume_create(&storage), "create content-reuse volume");
+ },
+ );
+ let contents = deterministic_bytes(8 * 1024 * 1024, 37);
+ fs::write(replica_a.join("original.bin"), &contents).expect("write original content");
+ fixture.observe(
+ "content-reuse",
+ "publish original",
+ LogicalIo {
+ read_bytes: 0,
+ write_bytes: contents.len() as u64,
+ },
+ || {
+ require_success(
+ ofs.sync(&replica_a, &state_a, &storage),
+ "publish original content",
+ );
+ },
+ );
+ let data_segments =
+ "local/managed-sync/content-reuse/managed/0/objects/00000000000000000000/04-data-segment";
+ let initial_encoded_bytes = fixture.storage_usage(data_segments).1;
+
+ fs::write(replica_a.join("copy.bin"), &contents).expect("write duplicate content");
+ fixture.observe(
+ "content-reuse",
+ "publish duplicate",
+ LogicalIo {
+ read_bytes: 0,
+ write_bytes: contents.len() as u64,
+ },
+ || {
+ require_success(
+ ofs.sync(&replica_a, &state_a, &storage),
+ "publish duplicate content",
+ );
+ },
+ );
+ let duplicated_encoded_bytes = fixture.storage_usage(data_segments).1;
+ assert_eq!(
+ duplicated_encoded_bytes, initial_encoded_bytes,
+ "content already reachable from the authority is reused across paths"
+ );
+
+ fixture.observe(
+ "content-reuse",
+ "cold restore",
+ LogicalIo {
+ read_bytes: 2 * contents.len() as u64,
+ write_bytes: 0,
+ },
+ || {
+ require_success(
+ ofs.sync(&replica_b, &state_b, &storage),
+ "restore reused content",
+ );
+ },
+ );
+ assert_eq!(
+ tree_summary(&replica_a),
+ tree_summary(&replica_b),
+ "reused content restores to every namespace path"
+ );
+}
+
+pub(super) fn extensions(fixture: &Fixture, ofs: Ofs) {
+ let rejected = fixture.storage_url("extensions/zstd-only");
+ let rejected = require_failure(
+ ofs.volume_create_zstd_only(&rejected),
+ "reject Zstandard without a finite partitioner",
+ );
+ assert!(
+ output_text(&rejected.stderr).contains("finite maximum extent"),
+ "Zstandard without FastCDC is rejected before any mutation"
+ );
+
+ let root = CaseRoot::new();
+ let replica_a = root.path.join("replica-a");
+ let replica_b = root.path.join("replica-b");
+ let state_a = root.path.join("state-a");
+ let state_b = root.path.join("state-b");
+ fs::create_dir_all(&replica_a).expect("create extension replica A");
+ fs::create_dir_all(&replica_b).expect("create extension replica B");
+ let storage = fixture.storage_url("extensions/fastcdc-zstd");
+
+ require_success(
+ ofs.volume_create_fastcdc_zstd(&storage),
+ "create extension volume",
+ );
+ fs::write(replica_a.join("first"), b"first extension file\n").expect("write extension file");
+ fs::write(replica_a.join("second"), b"second extension file\n").expect("write extension file");
+ require_success(
+ ofs.sync(&replica_a, &state_a, &storage),
+ "publish extension files",
+ );
+ require_success(
+ ofs.sync(&replica_b, &state_b, &storage),
+ "restore extension files",
+ );
+ fs::write(replica_b.join("first"), b"updated extension file\n").expect("update extension file");
+ require_success(
+ ofs.sync(&replica_b, &state_b, &storage),
+ "publish extension update",
+ );
+ require_success(
+ ofs.sync(&replica_a, &state_a, &storage),
+ "install extension update",
+ );
+ require_success(ofs.gc(&storage), "collect extension volume");
+ assert_eq!(
+ tree_summary(&replica_a),
+ tree_summary(&replica_b),
+ "extensions preserve the visible file tree"
+ );
+}
+
+pub(super) fn branch(fixture: &Fixture, ofs: Ofs) {
+ let root = CaseRoot::new();
+ let main = root.path.join("main");
+ let feature_a = root.path.join("feature-a");
+ let feature_b = root.path.join("feature-b");
+ let main_state = root.path.join("main-state");
+ let feature_a_state = root.path.join("feature-a-state");
+ let feature_b_state = root.path.join("feature-b-state");
+ for replica in [&main, &feature_a, &feature_b] {
+ fs::create_dir_all(replica).expect("create branch replica");
+ }
+ let storage = fixture.storage_url("branch");
+
+ require_success(ofs.volume_create_branch(&storage), "create branch volume");
+ fs::write(main.join("shared.txt"), b"main\n").expect("write main file");
+ require_success(
+ ofs.sync_branch(&main, &main_state, &storage, "main"),
+ "publish main branch",
+ );
+ require_success(
+ ofs.branch_create(&storage, "feature", "main"),
+ "create feature branch",
+ );
+ require_success(
+ ofs.sync_branch(&feature_a, &feature_a_state, &storage, "feature"),
+ "cold-install feature branch",
+ );
+ fs::write(feature_a.join("shared.txt"), b"feature\n").expect("change feature file");
+ fs::write(feature_a.join("feature.txt"), b"isolated\n").expect("add feature file");
+ require_success(
+ ofs.sync_branch(&feature_a, &feature_a_state, &storage, "feature"),
+ "publish feature branch",
+ );
+ require_success(
+ ofs.sync_branch(&feature_b, &feature_b_state, &storage, "feature"),
+ "install feature branch on a second replica",
+ );
+ assert_eq!(
+ tree_summary(&feature_a),
+ tree_summary(&feature_b),
+ "two replicas of one branch converge"
+ );
+ require_success(
+ ofs.sync_branch(&main, &main_state, &storage, "main"),
+ "observe isolated main branch",
+ );
+ assert_eq!(
+ fs::read(main.join("shared.txt")).expect("read main file"),
+ b"main\n",
+ "feature publication does not change main"
+ );
+ assert!(
+ !main.join("feature.txt").exists(),
+ "feature-only paths remain isolated"
+ );
+
+ require_success(ofs.gc(&storage), "collect branch volume");
+ require_success(
+ ofs.sync_branch(&feature_b, &feature_b_state, &storage, "feature"),
+ "read feature branch after collection",
+ );
+ let branches = output_text(&require_success(ofs.branch_list(&storage), "list branches").stdout);
+ assert_eq!(branches, "feature\nmain", "branch list is name ordered");
+ require_success(
+ ofs.branch_delete(&storage, "feature"),
+ "delete feature branch",
+ );
+ require_success(ofs.gc(&storage), "collect deleted branch");
+ let branches =
+ output_text(&require_success(ofs.branch_list(&storage), "list remaining branches").stdout);
+ assert_eq!(
+ branches, "main",
+ "deleted branches are no longer selectable"
+ );
+}
+
+pub(super) fn admission(fixture: &Fixture, ofs: Ofs) {
+ let root = CaseRoot::new();
+ let replica_a = root.path.join("replica-a");
+ let replica_b = root.path.join("replica-b");
+ let state_a = root.path.join("state-a");
+ let state_b = root.path.join("state-b");
+ fs::create_dir_all(&replica_a).expect("create replica A");
+ fs::create_dir_all(&replica_b).expect("create replica B");
+
+ let storage_a = fixture.storage_url("admission/a");
+ let storage_b = fixture.storage_url("admission/b");
+ let initialized_a = require_success(ofs.volume_create(&storage_a), "create volume A");
+ let volume_a = output_text(&initialized_a.stdout)
+ .split_whitespace()
+ .last()
+ .expect("initialization reports its volume identity")
+ .to_owned();
+ let mut attach_a = ofs.sync(&replica_a, &state_a, &storage_a);
+ attach_a.args(["--require", "portable-names"]);
+ require_success(attach_a, "attach replica A with its required capability");
+
+ let mut unsupported = ofs.sync(&replica_a, &state_a, &storage_a);
+ unsupported.args(["--require", "hard-link"]);
+ let unsupported = require_failure(unsupported, "require unsupported hard links");
+ assert!(
+ output_text(&unsupported.stderr).contains("required filesystem capability is unavailable"),
+ "an unavailable required capability is rejected before synchronization"
+ );
+
+ let status = ManagedStatus::parse(output_text(
+ &require_success(ofs.status(&replica_a, &state_a), "read replica status").stdout,
+ ));
+ assert_eq!(
+ status.volume_id, volume_a,
+ "status reports the initialized remote identity"
+ );
+ assert!(
+ !status.capabilities.extended_attributes,
+ "status reports that extended attributes are unavailable"
+ );
+ assert!(
+ status.capabilities.portable_names,
+ "status advertises the portable-name contract"
+ );
+
+ let portable_path = "资料/café.txt";
+ fs::create_dir_all(replica_a.join("资料")).expect("create portable Unicode directory");
+ fs::write(replica_a.join(portable_path), b"portable name\n")
+ .expect("write portable Unicode file");
+ require_success(
+ ofs.sync(&replica_a, &state_a, &storage_a),
+ "publish portable Unicode name",
+ );
+ require_success(
+ ofs.sync(&replica_b, &state_b, &storage_a),
+ "install portable Unicode name in replica B",
+ );
+ assert_eq!(
+ tree_summary(&replica_a),
+ tree_summary(&replica_b),
+ "a valid NFC name converges across replicas"
+ );
+ let portable_sequence = ManagedStatus::parse(output_text(
+ &require_success(
+ ofs.status(&replica_b, &state_b),
+ "read portable-name sequence",
+ )
+ .stdout,
+ ))
+ .remote_sequence;
+
+ for (names, expected_error, action) in [
+ (
+ &["Case.txt", "case.txt"][..],
+ "case-folding collision",
+ "reject a case-folding collision",
+ ),
+ (
+ &["cafe\u{301}.txt"][..],
+ "path component is not portable",
+ "reject a non-NFC component",
+ ),
+ (
+ &["CON.txt"][..],
+ "path component is reserved",
+ "reject a platform-reserved name",
+ ),
+ ] {
+ for name in names {
+ fs::write(replica_a.join(name), b"invalid portable name\n")
+ .expect("write rejected portable name");
+ }
+ let rejected = require_failure(ofs.sync(&replica_a, &state_a, &storage_a), action);
+ assert!(
+ output_text(&rejected.stderr).contains(expected_error),
+ "the rejected name reports the violated portable-name contract"
+ );
+ for name in names {
+ fs::remove_file(replica_a.join(name)).expect("remove rejected portable name");
+ }
+ let unchanged = require_success(
+ ofs.sync(&replica_b, &state_b, &storage_a),
+ "observe remote after rejected portable name",
+ );
+ assert!(
+ !output_text(&unchanged.stdout).contains("(published)"),
+ "a rejected portable name does not publish remote changes"
+ );
+ let remote = ManagedStatus::parse(output_text(
+ &require_success(
+ ofs.status(&replica_b, &state_b),
+ "read sequence after rejected portable name",
+ )
+ .stdout,
+ ));
+ assert_eq!(
+ remote.remote_sequence, portable_sequence,
+ "a rejected portable name does not advance the remote sequence"
+ );
+ }
+
+ require_success(ofs.volume_create(&storage_b), "create volume B");
+ let fenced = require_failure(
+ ofs.sync(&replica_a, &state_a, &storage_b),
+ "open replica A against volume B",
+ );
+ assert!(
+ output_text(&fenced.stderr).contains("different volume"),
+ "volume mismatch is reported at the command boundary"
+ );
+
+ for state in [&state_a, &state_b] {
+ let bytes = fs::read(state).expect("read behavior state");
+ assert!(
+ !bytes
+ .windows(b"minioadmin".len())
+ .any(|part| part == b"minioadmin"),
+ "provider credentials are not persisted in replica state"
+ );
+ assert!(
+ !bytes
+ .windows(storage_a.len())
+ .any(|part| part == storage_a.as_bytes())
+ && !bytes
+ .windows(storage_b.len())
+ .any(|part| part == storage_b.as_bytes()),
+ "storage configuration is not persisted in replica state"
+ );
+ }
+}
+
+pub(super) fn smoke(fixture: &Fixture, ofs: Ofs) {
+ let root = CaseRoot::new();
+ let replica_a = root.path.join("replica-a");
+ let replica_b = root.path.join("replica-b");
+ let state_a = root.path.join("state-a");
+ let state_b = root.path.join("state-b");
+ fs::create_dir_all(replica_a.join("nested")).expect("create replica A");
+ fs::create_dir_all(&replica_b).expect("create replica B");
+ let storage = fixture.storage_url("smoke");
+
+ fixture.observe("smoke", "create volume", LogicalIo::default(), || {
+ require_success(ofs.volume_create(&storage), "create smoke volume");
+ });
+ fs::write(replica_a.join("empty"), []).expect("write empty file");
+ fs::write(replica_a.join("nested/one"), b"shared content\n").expect("write nested file");
+ fs::write(replica_a.join("two"), b"shared content\n").expect("write repeated file");
+ fs::write(replica_a.join("tool"), b"#!/bin/sh\nexit 0\n").expect("write executable file");
+ make_executable(&replica_a.join("tool"));
+
+ let initial_logical_bytes = ["empty", "nested/one", "two", "tool"]
+ .into_iter()
+ .map(|path| {
+ fs::metadata(replica_a.join(path))
+ .expect("read smoke file metadata")
+ .len()
+ })
+ .sum();
+ let published = fixture.observe(
+ "smoke",
+ "initial publish",
+ LogicalIo {
+ read_bytes: 0,
+ write_bytes: initial_logical_bytes,
+ },
+ || {
+ require_success(
+ ofs.sync(&replica_a, &state_a, &storage),
+ "publish smoke tree",
+ )
+ },
+ );
+ assert!(
+ output_text(&published.stdout).contains("(published)"),
+ "a changed tree reports remote publication"
+ );
+ fixture.observe(
+ "smoke",
+ "cold restore",
+ LogicalIo {
+ read_bytes: initial_logical_bytes,
+ write_bytes: 0,
+ },
+ || {
+ require_success(
+ ofs.sync(&replica_b, &state_b, &storage),
+ "cold restore smoke tree",
+ );
+ },
+ );
+ assert_eq!(
+ tree_summary(&replica_a),
+ tree_summary(&replica_b),
+ "cold restore reproduces files, directories, content, and executable state"
+ );
+
+ let no_op = fixture.observe("smoke", "no-op sync", LogicalIo::default(), || {
+ require_success(
+ ofs.sync(&replica_a, &state_a, &storage),
+ "repeat unchanged sync",
+ )
+ });
+ assert!(
+ !output_text(&no_op.stdout).contains("(published)"),
+ "an unchanged sync does not publish a namespace generation"
+ );
+
+ fs::write(replica_a.join("nested/one"), b"changed content\n").expect("change nested file");
+ let changed_bytes = fs::metadata(replica_a.join("nested/one"))
+ .expect("read changed smoke file metadata")
+ .len();
+ fixture.observe(
+ "smoke",
+ "publish update",
+ LogicalIo {
+ read_bytes: 0,
+ write_bytes: changed_bytes,
+ },
+ || {
+ require_success(
+ ofs.sync(&replica_a, &state_a, &storage),
+ "publish changed smoke tree",
+ );
+ },
+ );
+ fixture.observe(
+ "smoke",
+ "install update",
+ LogicalIo {
+ read_bytes: changed_bytes,
+ write_bytes: 0,
+ },
+ || {
+ require_success(
+ ofs.sync(&replica_b, &state_b, &storage),
+ "install changed smoke tree",
+ );
+ },
+ );
+ assert_eq!(
+ tree_summary(&replica_a),
+ tree_summary(&replica_b),
+ "a later remote generation converges into the peer replica"
+ );
+}
diff --git a/xtask/src/managed_sync/behavior/reconciliation.rs b/xtask/src/managed_sync/behavior/reconciliation.rs
new file mode 100644
index 0000000..da23bbf
--- /dev/null
+++ b/xtask/src/managed_sync/behavior/reconciliation.rs
@@ -0,0 +1,229 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Multi-replica reconciliation and rename behavior.
+
+use std::fs;
+
+use super::super::cli::{ManagedStatus, Ofs, output_text, require_failure, require_success};
+use super::super::fixture::{CaseRoot, Fixture, tree_summary};
+use super::make_executable;
+
+pub(super) fn rename(fixture: &Fixture, ofs: Ofs) {
+ let root = CaseRoot::new();
+ let replica_a = root.path.join("replica-a");
+ let replica_b = root.path.join("replica-b");
+ let state_a = root.path.join("state-a");
+ let state_b = root.path.join("state-b");
+ fs::create_dir_all(replica_a.join("tree-before/subtree/empty"))
+ .expect("create rename source tree");
+ fs::create_dir_all(&replica_b).expect("create rename replica B");
+ fs::write(replica_a.join("file-before"), b"stable file\n").expect("write rename file");
+ fs::write(
+ replica_a.join("tree-before/subtree/leaf"),
+ b"stable directory tree\n",
+ )
+ .expect("write rename tree leaf");
+ let storage = fixture.storage_url("rename");
+
+ require_success(ofs.volume_create(&storage), "create rename volume");
+ require_success(
+ ofs.sync(&replica_a, &state_a, &storage),
+ "publish rename base",
+ );
+ require_success(
+ ofs.sync(&replica_b, &state_b, &storage),
+ "attach rename replica B",
+ );
+
+ fs::rename(replica_a.join("file-before"), replica_a.join("file-after")).expect("rename file");
+ fs::rename(replica_a.join("tree-before"), replica_a.join("tree-after"))
+ .expect("move directory tree");
+ make_executable(&replica_a.join("file-after"));
+ require_success(
+ ofs.sync(&replica_a, &state_a, &storage),
+ "publish file and directory moves",
+ );
+
+ require_success(
+ ofs.sync(&replica_b, &state_b, &storage),
+ "install remote moves",
+ );
+ assert!(
+ !replica_b.join("file-before").exists() && !replica_b.join("tree-before").exists(),
+ "old paths disappear after remote moves"
+ );
+ assert!(
+ replica_b.join("tree-after/subtree/empty").is_dir(),
+ "a moved empty directory is retained"
+ );
+ assert_eq!(
+ tree_summary(&replica_a),
+ tree_summary(&replica_b),
+ "file identity, directory contents, and attributes survive moves"
+ );
+
+ fs::write(replica_b.join("file-after"), b"edited after move\n")
+ .expect("edit moved file in peer");
+ require_success(
+ ofs.sync(&replica_b, &state_b, &storage),
+ "publish edit through moved identity",
+ );
+ require_success(
+ ofs.sync(&replica_a, &state_a, &storage),
+ "install post-move peer edit",
+ );
+ assert_eq!(
+ fs::read(replica_a.join("file-after")).expect("read post-move edit"),
+ b"edited after move\n"
+ );
+}
+
+pub(super) fn reconcile(fixture: &Fixture, ofs: Ofs) {
+ let root = CaseRoot::new();
+ let replica_a = root.path.join("replica-a");
+ let replica_b = root.path.join("replica-b");
+ let state_a = root.path.join("state-a");
+ let state_b = root.path.join("state-b");
+ fs::create_dir_all(&replica_a).expect("create reconcile replica A");
+ fs::create_dir_all(&replica_b).expect("create reconcile replica B");
+ let storage = fixture.storage_url("reconcile");
+
+ require_success(ofs.volume_create(&storage), "create reconcile volume");
+ fs::write(replica_a.join("shared.txt"), b"common\n").expect("write common file");
+ fs::write(replica_a.join("delete-edit.txt"), b"common\n").expect("write delete-edit base");
+ require_success(
+ ofs.sync(&replica_a, &state_a, &storage),
+ "publish reconcile base",
+ );
+ require_success(
+ ofs.sync(&replica_b, &state_b, &storage),
+ "attach reconcile replica B",
+ );
+
+ fs::write(replica_a.join("from-a.txt"), b"from A\n").expect("write A-only change");
+ fs::write(replica_b.join("from-b.txt"), b"from B\n").expect("write B-only change");
+ require_success(
+ ofs.sync(&replica_a, &state_a, &storage),
+ "publish A-only change",
+ );
+ let merged = require_success(
+ ofs.sync(&replica_b, &state_b, &storage),
+ "merge B-only change",
+ );
+ assert!(
+ output_text(&merged.stdout).contains("(published)"),
+ "a disjoint two-replica merge publishes one combined generation"
+ );
+ let data_segments =
+ "local/managed-sync/reconcile/managed/0/objects/00000000000000000000/04-data-segment";
+ let data_objects = fixture.storage_usage(data_segments).0;
+ require_success(
+ ofs.sync(&replica_a, &state_a, &storage),
+ "install disjoint merge in A",
+ );
+ assert_eq!(
+ fixture.storage_usage(data_segments).0,
+ data_objects,
+ "installing a merged remote generation does not publish unreachable data"
+ );
+ assert_eq!(
+ tree_summary(&replica_a),
+ tree_summary(&replica_b),
+ "disjoint changes from both replicas converge"
+ );
+
+ fs::write(replica_a.join("shared.txt"), b"candidate A\n").expect("write A candidate");
+ fs::write(replica_b.join("shared.txt"), b"candidate B\n").expect("write B candidate");
+ require_success(
+ ofs.sync(&replica_a, &state_a, &storage),
+ "publish A conflict candidate",
+ );
+ let conflict = require_failure(
+ ofs.sync(&replica_b, &state_b, &storage),
+ "retain concurrent file conflict",
+ );
+ let conflict_message = output_text(&conflict.stderr);
+ assert!(
+ conflict_message.contains("retained 1 conflict")
+ && conflict_message.contains("--resolve <relative-path>"),
+ "a concurrent file update explains explicit resolution"
+ );
+ let conflict_path = conflict_message
+ .lines()
+ .find_map(|line| line.strip_prefix(" "))
+ .expect("a concurrent file update reports its normalized relative path");
+ assert_eq!(
+ conflict_path, "shared.txt",
+ "the reported conflict identifies the user-visible path"
+ );
+ assert_eq!(
+ fs::read(replica_a.join("shared.txt")).expect("read remote candidate"),
+ b"candidate A\n"
+ );
+ assert_eq!(
+ fs::read(replica_b.join("shared.txt")).expect("read local candidate"),
+ b"candidate B\n"
+ );
+ let status = ManagedStatus::parse(output_text(
+ &require_success(ofs.status(&replica_b, &state_b), "report retained conflict").stdout,
+ ));
+ assert_eq!(
+ status.conflicts, 1,
+ "status reports the unresolved conflict count"
+ );
+ require_success(
+ ofs.sync_resolve(&replica_b, &state_b, &storage, &[conflict_path]),
+ "resolve file conflict with local candidate",
+ );
+ require_success(
+ ofs.sync(&replica_a, &state_a, &storage),
+ "install resolved file in A",
+ );
+ assert_eq!(
+ fs::read(replica_a.join("shared.txt")).expect("read resolved candidate"),
+ b"candidate B\n",
+ "explicit resolution publishes the selected local content"
+ );
+
+ fs::write(replica_a.join("delete-edit.txt"), b"edited in A\n").expect("edit delete-edit file");
+ fs::remove_file(replica_b.join("delete-edit.txt")).expect("delete file in B");
+ require_success(
+ ofs.sync(&replica_a, &state_a, &storage),
+ "publish edit before delete conflict",
+ );
+ require_failure(
+ ofs.sync(&replica_b, &state_b, &storage),
+ "retain delete-versus-edit conflict",
+ );
+ assert!(
+ replica_a.join("delete-edit.txt").is_file() && !replica_b.join("delete-edit.txt").exists(),
+ "delete-versus-edit retains both available user outcomes"
+ );
+ require_success(
+ ofs.sync_resolve(&replica_b, &state_b, &storage, &["delete-edit.txt"]),
+ "resolve delete-versus-edit with local deletion",
+ );
+ require_success(
+ ofs.sync(&replica_a, &state_a, &storage),
+ "install resolved deletion in A",
+ );
+ assert!(
+ !replica_a.join("delete-edit.txt").exists(),
+ "explicit local deletion resolution converges"
+ );
+}
diff --git a/xtask/src/managed_sync/behavior/recovery.rs b/xtask/src/managed_sync/behavior/recovery.rs
new file mode 100644
index 0000000..b59922d
--- /dev/null
+++ b/xtask/src/managed_sync/behavior/recovery.rs
@@ -0,0 +1,144 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Collection and process-interruption behavior.
+
+use std::fs;
+use std::path::Path;
+use std::process::Command;
+use std::thread;
+use std::time::{Duration, Instant};
+
+use super::super::cli::{ManagedStatus, Ofs, output_text, require_failure, require_success};
+use super::super::fixture::{CaseRoot, Fixture, tree_summary};
+
+pub(super) fn gc(fixture: &Fixture, ofs: Ofs) {
+ let root = CaseRoot::new();
+ let a = root.path.join("a");
+ let b = root.path.join("b");
+ let state_a = root.path.join("state-a");
+ let state_b = root.path.join("state-b");
+ fs::create_dir_all(&a).expect("create GC source");
+ fs::create_dir_all(&b).expect("create GC destination");
+ let storage = fixture.storage_url("gc");
+
+ require_success(ofs.volume_create(&storage), "create GC volume");
+ fs::write(a.join("changing.bin"), b"superseded\n").expect("write initial GC content");
+ fs::write(a.join("live.bin"), b"live\n").expect("write live GC content");
+ require_success(ofs.sync(&a, &state_a, &storage), "publish initial GC tree");
+ fs::write(a.join("changing.bin"), b"current\n").expect("replace GC content");
+ require_success(ofs.sync(&a, &state_a, &storage), "publish replacement");
+ require_success(ofs.gc(&storage), "collect unreachable objects");
+ require_success(ofs.sync(&b, &state_b, &storage), "restore after collection");
+ assert_eq!(tree_summary(&a), tree_summary(&b));
+ require_success(ofs.gc(&storage), "repeat completed collection");
+}
+
+pub(super) fn interruption(fixture: &Fixture, ofs: Ofs) {
+ let root = CaseRoot::new();
+ let a = root.path.join("a");
+ let b = root.path.join("b");
+ let state_a = root.path.join("state-a");
+ let state_b = root.path.join("state-b");
+ fs::create_dir_all(&a).expect("create interrupted source");
+ fs::create_dir_all(&b).expect("create interrupted destination");
+ fs::write(a.join("file"), b"interrupted publication\n").expect("write interrupted content");
+ let storage = fixture.storage_url("interruption");
+ require_success(ofs.volume_create(&storage), "create interruption volume");
+
+ interrupt(ofs.sync(&a, &state_a, &storage), &state_a);
+ require_success(
+ ofs.sync(&a, &state_a, &storage),
+ "retry interrupted publish",
+ );
+ interrupt(ofs.sync(&b, &state_b, &storage), &state_b);
+ require_success(
+ ofs.sync(&b, &state_b, &storage),
+ "retry interrupted restore",
+ );
+ assert_eq!(tree_summary(&a), tree_summary(&b));
+}
+
+fn interrupt(mut command: Command, progress: &Path) {
+ let mut child = command.spawn().expect("start interruptible OFS process");
+ let deadline = Instant::now() + Duration::from_secs(10);
+ while !progress.exists() {
+ assert!(
+ child.try_wait().expect("inspect OFS process").is_none(),
+ "OFS completed before the interruption point"
+ );
+ assert!(Instant::now() < deadline, "OFS did not expose progress");
+ thread::sleep(Duration::from_millis(1));
+ }
+ child.kill().expect("interrupt OFS process");
+ assert!(!child.wait().expect("wait for interrupted OFS").success());
+}
+
+pub(super) fn offline_gc(fixture: &Fixture, ofs: Ofs) {
+ let root = CaseRoot::new();
+ let a = root.path.join("a");
+ let b = root.path.join("b");
+ let expired = root.path.join("expired");
+ let restored = root.path.join("restored");
+ let state_a = root.path.join("state-a");
+ let state_b = root.path.join("state-b");
+ let state_expired = root.path.join("state-expired");
+ let state_restored = root.path.join("state-restored");
+ for replica in [&a, &b, &expired, &restored] {
+ fs::create_dir_all(replica).expect("create GC replica");
+ }
+ let storage = fixture.storage_url("offline-gc");
+ require_success(ofs.volume_create(&storage), "create offline GC volume");
+
+ fs::write(a.join("cursor.txt"), b"first base\n").expect("write first base");
+ require_success(ofs.sync(&a, &state_a, &storage), "publish first base");
+ require_success(
+ ofs.sync(&expired, &state_expired, &storage),
+ "attach replica before collection",
+ );
+ fs::write(a.join("cursor.txt"), b"collection base\n").expect("write collection base");
+ require_success(ofs.sync(&a, &state_a, &storage), "publish collection base");
+ require_success(ofs.sync(&b, &state_b, &storage), "attach current replica");
+ require_success(ofs.gc(&storage), "collect from current namespace");
+
+ fs::write(a.join("remote.txt"), b"remote\n").expect("write remote change");
+ require_success(ofs.sync(&a, &state_a, &storage), "publish remote change");
+ fs::write(b.join("local.txt"), b"local\n").expect("write local change");
+ require_success(ofs.sync(&b, &state_b, &storage), "merge current replica");
+ require_success(ofs.sync(&a, &state_a, &storage), "converge current replica");
+ assert_eq!(tree_summary(&a), tree_summary(&b));
+
+ fs::write(expired.join("cursor.txt"), b"expired local change\n").expect("write expired change");
+ require_failure(
+ ofs.sync(&expired, &state_expired, &storage),
+ "reject an unavailable base",
+ );
+ let status = ManagedStatus::parse(output_text(
+ &require_success(
+ ofs.status(&expired, &state_expired),
+ "read unavailable-base status",
+ )
+ .stdout,
+ ));
+ assert!(status.base_expired && status.conflicts != 0 && !status.pending);
+
+ require_success(
+ ofs.sync(&restored, &state_restored, &storage),
+ "restore while an expired replica remains unresolved",
+ );
+ assert_eq!(tree_summary(&a), tree_summary(&restored));
+}
diff --git a/xtask/src/managed_sync/bub.rs b/xtask/src/managed_sync/bub.rs
new file mode 100644
index 0000000..259f4d4
--- /dev/null
+++ b/xtask/src/managed_sync/bub.rs
@@ -0,0 +1,497 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Bub end-to-end scenario across isolated containers.
+
+use std::env;
+use std::fs;
+use std::path::Path;
+use std::process::Output;
+
+use super::cli::{ManagedStatus, output_text, volume_name};
+use super::fixture::{CaseRoot, Fixture, container_http_proxy, tree_summary};
+
+pub(crate) fn run(keep: bool) {
+ let api_key = env::var("BUB_API_KEY")
+ .ok()
+ .filter(|value| !value.is_empty())
+ .expect("BUB_API_KEY is required for the Bub end-to-end scenario");
+ let fixture = Fixture::new(keep, "bub", Default::default()).start_bub();
+ fixture.create_bucket();
+ let sessions = fixture.storage_url("bub/sessions");
+ let skills = fixture.storage_url("bub/skills");
+ let mut observed_output = Vec::new();
+
+ for service in ["bub-a", "bub-b"] {
+ container_success(
+ &fixture,
+ service,
+ &[
+ "mkdir",
+ "-p",
+ "/sync/sessions/tapes",
+ "/workspace/.agents/skills",
+ "/var/lib/ofs",
+ "/var/lib/bub/.bub",
+ ],
+ "prepare isolated Bub client",
+ );
+ container_success(
+ &fixture,
+ service,
+ &["ofs", "--version"],
+ "verify installed OFS",
+ );
+ container_success(
+ &fixture,
+ service,
+ &["bub", "--help"],
+ "verify installed Bub",
+ );
+ }
+ container_success(
+ &fixture,
+ "bub-a",
+ &["touch", "/var/lib/ofs/a-only"],
+ "create client A isolation marker",
+ );
+ container_success(
+ &fixture,
+ "bub-b",
+ &["test", "!", "-e", "/var/lib/ofs/a-only"],
+ "verify client filesystem isolation",
+ );
+ container_success(
+ &fixture,
+ "bub-b",
+ &["touch", "/var/lib/ofs/b-only"],
+ "create client B isolation marker",
+ );
+ container_success(
+ &fixture,
+ "bub-a",
+ &["test", "!", "-e", "/var/lib/ofs/b-only"],
+ "verify reverse client filesystem isolation",
+ );
+
+ for (replica, state, storage) in [
+ (
+ "/sync/sessions/tapes",
+ "/var/lib/ofs/sessions.state",
+ &sessions,
+ ),
+ (
+ "/workspace/.agents/skills",
+ "/var/lib/ofs/skills.state",
+ &skills,
+ ),
+ ] {
+ container_volume_create(&fixture, "bub-a", storage);
+ container_sync(&fixture, "bub-b", replica, state, storage, &[]);
+ }
+
+ let fact_a = format!("HARBOR-A-{}", std::process::id());
+ observed_output.push(container_bub(
+ &fixture,
+ "bub-a",
+ "handoff-a",
+ &format!("Remember this exact handoff fact for later: {fact_a}. Reply with the fact."),
+ ));
+ let marker_a = format!("BUB-SKILL-A-{}", std::process::id());
+ observed_output.push(container_bub(
+ &fixture,
+ "bub-a",
+ "create-skill-a",
+ &skill_creation_prompt("sync-a", &marker_a),
+ ));
+ container_success(
+ &fixture,
+ "bub-a",
+ &["test", "-f", "/workspace/.agents/skills/sync-a/SKILL.md"],
+ "verify Bub created skill A",
+ );
+ container_success(
+ &fixture,
+ "bub-b",
+ &["test", "!", "-e", "/workspace/.agents/skills/sync-a"],
+ "verify skill A is isolated before synchronization",
+ );
+ converge(&fixture, "bub-a", "bub-b", &sessions, &skills);
+
+ let continued_a = container_bub(
+ &fixture,
+ "bub-b",
+ "handoff-a",
+ "Return the exact handoff fact that I previously asked you to remember.",
+ );
+ assert!(
+ continued_a.contains(&fact_a),
+ "Bub session created in A did not continue in B"
+ );
+ observed_output.push(continued_a);
+ let invoked_a = container_bub(
+ &fixture,
+ "bub-b",
+ "invoke-skill-a",
+ "$sync-a Follow this project skill now.",
+ );
+ assert!(
+ invoked_a.contains(&marker_a),
+ "Bub skill created in A was not discovered and invoked in B"
+ );
+ observed_output.push(invoked_a);
+
+ let fact_b = format!("HARBOR-B-{}", std::process::id());
+ observed_output.push(container_bub(
+ &fixture,
+ "bub-b",
+ "handoff-b",
+ &format!("Remember this exact handoff fact for later: {fact_b}. Reply with the fact."),
+ ));
+ let marker_b = format!("BUB-SKILL-B-{}", std::process::id());
+ observed_output.push(container_bub(
+ &fixture,
+ "bub-b",
+ "create-skill-b",
+ &skill_creation_prompt("sync-b", &marker_b),
+ ));
+ container_success(
+ &fixture,
+ "bub-b",
+ &["test", "-f", "/workspace/.agents/skills/sync-b/SKILL.md"],
+ "verify Bub created skill B",
+ );
+ container_success(
+ &fixture,
+ "bub-a",
+ &["test", "!", "-e", "/workspace/.agents/skills/sync-b"],
+ "verify skill B is isolated before synchronization",
+ );
+ converge(&fixture, "bub-b", "bub-a", &sessions, &skills);
+
+ let continued_b = container_bub(
+ &fixture,
+ "bub-a",
+ "handoff-b",
+ "Return the exact handoff fact that I previously asked you to remember.",
+ );
+ assert!(
+ continued_b.contains(&fact_b),
+ "Bub session created in B did not continue in A"
+ );
+ observed_output.push(continued_b);
+ let invoked_b = container_bub(
+ &fixture,
+ "bub-a",
+ "invoke-skill-b",
+ "$sync-b Follow this project skill now.",
+ );
+ assert!(
+ invoked_b.contains(&marker_b),
+ "Bub skill created in B was not discovered and invoked in A"
+ );
+ observed_output.push(invoked_b);
+
+ converge(&fixture, "bub-a", "bub-b", &sessions, &skills);
+ let before_noop = bub_statuses(&fixture);
+ for (replica, state, storage) in [
+ (
+ "/sync/sessions/tapes",
+ "/var/lib/ofs/sessions.state",
+ &sessions,
+ ),
+ (
+ "/workspace/.agents/skills",
+ "/var/lib/ofs/skills.state",
+ &skills,
+ ),
+ ] {
+ for service in ["bub-a", "bub-b"] {
+ let output = container_sync(&fixture, service, replica, state, storage, &[]);
+ assert!(
+ !output.contains("(published)"),
+ "final Bub convergence is not a no-op"
+ );
+ observed_output.push(output);
+ }
+ }
+
+ let statuses = bub_statuses(&fixture);
+ for (before, after) in before_noop.iter().zip(&statuses) {
+ assert_eq!(
+ before.common_sequence, after.common_sequence,
+ "final no-op advanced a common cursor"
+ );
+ assert_eq!(
+ before.remote_sequence, after.remote_sequence,
+ "final no-op advanced a remote cursor"
+ );
+ }
+ for status in &statuses {
+ assert!(
+ !status.pending && status.conflicts == 0,
+ "Bub convergence status contains pending work or conflicts"
+ );
+ assert_eq!(
+ status.common_sequence, status.remote_sequence,
+ "Bub replica stopped behind the remote cursor"
+ );
+ }
+ assert_eq!(
+ statuses[0].common_sequence, statuses[1].common_sequence,
+ "Bub session replicas ended at different cursors"
+ );
+ assert_eq!(
+ statuses[2].common_sequence, statuses[3].common_sequence,
+ "Bub skill replicas ended at different cursors"
+ );
+ observed_output.extend(statuses.into_iter().map(|status| status.document));
+
+ let evidence = CaseRoot::new();
+ for (service, suffix) in [("bub-a", "a"), ("bub-b", "b")] {
+ fixture.copy_from_container(
+ service,
+ "/sync/sessions/tapes/.",
+ &evidence.path.join(format!("sessions-{suffix}")),
+ );
+ fixture.copy_from_container(
+ service,
+ "/workspace/.agents/skills/.",
+ &evidence.path.join(format!("skills-{suffix}")),
+ );
+ fixture.copy_from_container(
+ service,
+ "/var/lib/ofs/.",
+ &evidence.path.join(format!("state-{suffix}")),
+ );
+ }
+ assert_eq!(
+ tree_summary(&evidence.path.join("sessions-a")),
+ tree_summary(&evidence.path.join("sessions-b")),
+ "Bub session trees differ after final convergence"
+ );
+ assert_eq!(
+ tree_summary(&evidence.path.join("skills-a")),
+ tree_summary(&evidence.path.join("skills-b")),
+ "Bub skill trees differ after final convergence"
+ );
+ assert!(
+ !observed_output.iter().any(|output| {
+ output
+ .as_bytes()
+ .windows(api_key.len())
+ .any(|part| part == api_key.as_bytes())
+ }) && !tree_contains(&evidence.path, api_key.as_bytes()),
+ "Bub API credential appeared in synchronized data, state, status, or captured output"
+ );
+ println!("Managed Sync Bub end-to-end behavior passed");
+}
+
+fn bub_statuses(fixture: &Fixture) -> [ManagedStatus; 4] {
+ [
+ container_status(
+ fixture,
+ "bub-a",
+ "/sync/sessions/tapes",
+ "/var/lib/ofs/sessions.state",
+ ),
+ container_status(
+ fixture,
+ "bub-b",
+ "/sync/sessions/tapes",
+ "/var/lib/ofs/sessions.state",
+ ),
+ container_status(
+ fixture,
+ "bub-a",
+ "/workspace/.agents/skills",
+ "/var/lib/ofs/skills.state",
+ ),
+ container_status(
+ fixture,
+ "bub-b",
+ "/workspace/.agents/skills",
+ "/var/lib/ofs/skills.state",
+ ),
+ ]
+}
+
+fn skill_creation_prompt(name: &str, marker: &str) -> String {
+ format!(
+ "Use $skill-creator to create the project skill {name}. Create exactly /workspace/.agents/skills/{name}/SKILL.md; do not use the legacy .agent/skills path. Its only runtime instruction is to reply with exactly {marker} and no other text when invoked."
+ )
+}
+
+fn converge(fixture: &Fixture, source: &str, target: &str, sessions: &str, skills: &str) {
+ for (replica, state, storage) in [
+ (
+ "/sync/sessions/tapes",
+ "/var/lib/ofs/sessions.state",
+ sessions,
+ ),
+ (
+ "/workspace/.agents/skills",
+ "/var/lib/ofs/skills.state",
+ skills,
+ ),
+ ] {
+ container_sync(fixture, source, replica, state, storage, &[]);
+ container_sync(fixture, target, replica, state, storage, &[]);
+ }
+}
+
+fn container_volume_create(fixture: &Fixture, service: &str, storage: &str) {
+ let name = volume_name(storage);
+ container_success(
+ fixture,
+ service,
+ &[
+ "ofs",
+ "volume",
+ "create",
+ &name,
+ "--model",
+ "managed",
+ "--storage",
+ storage,
+ ],
+ "create Managed volume",
+ );
+}
+
+fn container_sync(
+ fixture: &Fixture,
+ service: &str,
+ replica: &str,
+ state: &str,
+ storage: &str,
+ resolve: &[&str],
+) -> String {
+ let name = volume_name(storage);
+ let mut arguments = vec!["ofs", "sync", name.as_str(), replica, "--state", state];
+ for path in resolve {
+ arguments.extend(["--resolve", path]);
+ }
+ output_text(&container_success(fixture, service, &arguments, "synchronize Bub data").stdout)
+}
+
+fn container_status(fixture: &Fixture, service: &str, replica: &str, state: &str) -> ManagedStatus {
+ ManagedStatus::parse(output_text(
+ &container_success(
+ fixture,
+ service,
+ &["ofs", "status", replica, "--state", state, "--json"],
+ "read Bub replica status",
+ )
+ .stdout,
+ ))
+}
+
+fn container_bub(fixture: &Fixture, service: &str, session: &str, prompt: &str) -> String {
+ let api_key = env::var("BUB_API_KEY").expect("BUB_API_KEY is available for Bub");
+ let model = env::var("BUB_MODEL").unwrap_or_else(|_| "deepseek:deepseek-chat".to_owned());
+ let proxy = container_http_proxy();
+ for attempt in 1..=3 {
+ let mut command = fixture.compose();
+ command
+ .env("BUB_API_KEY", &api_key)
+ .env("BUB_MODEL", &model)
+ .args(["exec", "-T", "-e", "BUB_API_KEY", "-e", "BUB_MODEL"]);
+ if let Some(proxy) = &proxy {
+ command
+ .env("HTTP_PROXY", proxy)
+ .env("HTTPS_PROXY", proxy)
+ .env("http_proxy", proxy)
+ .env("https_proxy", proxy)
+ .args([
+ "-e",
+ "HTTP_PROXY",
+ "-e",
+ "HTTPS_PROXY",
+ "-e",
+ "http_proxy",
+ "-e",
+ "https_proxy",
+ ]);
+ }
+ let output = command
+ .arg(service)
+ .args([
+ "bub",
+ "--workspace",
+ "/workspace",
+ "run",
+ prompt,
+ "--session-id",
+ session,
+ ])
+ .output()
+ .expect("execute Bub turn");
+ let transcript = format!(
+ "{}\n{}",
+ output_text(&output.stdout),
+ output_text(&output.stderr)
+ );
+ assert!(
+ !transcript.contains(&api_key),
+ "Bub API credential appeared in captured turn output"
+ );
+ if output.status.success() {
+ return transcript;
+ }
+ if attempt == 3 {
+ panic!(
+ "run Bub turn failed after {attempt} attempts ({}); output withheld",
+ output.status
+ );
+ }
+ }
+ unreachable!()
+}
+
+fn container_success(fixture: &Fixture, service: &str, arguments: &[&str], action: &str) -> Output {
+ let mut command = fixture.compose();
+ command.args(["exec", "-T", service]).args(arguments);
+ let output = command.output().expect("execute container command");
+ assert!(
+ output.status.success(),
+ "{action} failed: {}",
+ output_text(&output.stderr)
+ );
+ output
+}
+
+fn tree_contains(root: &Path, needle: &[u8]) -> bool {
+ let mut pending = vec![root.to_owned()];
+ while let Some(path) = pending.pop() {
+ let metadata = fs::metadata(&path).expect("read evidence metadata");
+ if metadata.is_dir() {
+ pending.extend(
+ fs::read_dir(path)
+ .expect("read evidence directory")
+ .map(|entry| entry.expect("read evidence entry").path()),
+ );
+ } else if fs::read(path)
+ .expect("read evidence file")
+ .windows(needle.len())
+ .any(|part| part == needle)
+ {
+ return true;
+ }
+ }
+ false
+}
diff --git a/xtask/src/managed_sync/cli.rs b/xtask/src/managed_sync/cli.rs
new file mode 100644
index 0000000..4274700
--- /dev/null
+++ b/xtask/src/managed_sync/cli.rs
@@ -0,0 +1,275 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! OFS command construction and process assertions.
+
+use std::env;
+use std::path::{Path, PathBuf};
+use std::process::{Command, Output};
+
+#[derive(Clone)]
+pub(super) struct Ofs {
+ release: bool,
+ transfer_concurrency: Option<usize>,
+ home: PathBuf,
+}
+
+impl Ofs {
+ pub(super) fn debug(home: impl Into<PathBuf>) -> Self {
+ Self {
+ release: false,
+ transfer_concurrency: None,
+ home: home.into(),
+ }
+ }
+
+ pub(super) fn release(home: impl Into<PathBuf>) -> Self {
+ Self {
+ release: true,
+ transfer_concurrency: Some(16),
+ home: home.into(),
+ }
+ }
+
+ pub(super) fn build(&self) {
+ let mut command = Command::new(env::var_os("CARGO").unwrap_or_else(|| "cargo".into()));
+ command
+ .current_dir(env!("CARGO_WORKSPACE_DIR"))
+ .arg("build");
+ if self.release {
+ command.arg("--release");
+ }
+ command.args(["--bin", "ofs"]);
+ run_logged(&mut command);
+ }
+
+ pub(super) fn volume_create(&self, storage: &str) -> Command {
+ let mut command = self.command();
+ command.args([
+ "volume",
+ "create",
+ &volume_name(storage),
+ "--model",
+ "managed",
+ "--storage",
+ storage,
+ ]);
+ command
+ }
+
+ pub(super) fn volume_create_fastcdc_zstd(&self, storage: &str) -> Command {
+ let mut command = self.volume_create(storage);
+ command.args(["--fastcdc", "--zstd-level", "3"]);
+ command
+ }
+
+ pub(super) fn volume_create_branch(&self, storage: &str) -> Command {
+ let mut command = self.volume_create(storage);
+ command.arg("--authority-branch");
+ command
+ }
+
+ pub(super) fn volume_create_zstd_only(&self, storage: &str) -> Command {
+ let mut command = self.volume_create(storage);
+ command.args(["--zstd-level", "3"]);
+ command
+ }
+
+ pub(super) fn branch_create(&self, storage: &str, name: &str, source: &str) -> Command {
+ let mut command = self.command();
+ command.args([
+ "branch",
+ &volume_name(storage),
+ "create",
+ name,
+ "--from",
+ source,
+ ]);
+ command
+ }
+
+ pub(super) fn branch_delete(&self, storage: &str, name: &str) -> Command {
+ let mut command = self.command();
+ command.args(["branch", &volume_name(storage), "delete", name]);
+ command
+ }
+
+ pub(super) fn branch_list(&self, storage: &str) -> Command {
+ let mut command = self.command();
+ command.args(["branch", &volume_name(storage), "list"]);
+ command
+ }
+
+ pub(super) fn sync_branch(
+ &self,
+ replica: &Path,
+ state: &Path,
+ storage: &str,
+ branch: &str,
+ ) -> Command {
+ let mut command = self.sync(replica, state, storage);
+ command.arg("--branch").arg(branch);
+ command
+ }
+
+ pub(super) fn sync(&self, replica: &Path, state: &Path, storage: &str) -> Command {
+ let mut command = self.command();
+ command
+ .arg("sync")
+ .arg(volume_name(storage))
+ .arg(replica)
+ .arg("--state")
+ .arg(state);
+ command
+ }
+
+ pub(super) fn sync_changes(
+ &self,
+ replica: &Path,
+ state: &Path,
+ storage: &str,
+ changes: &Path,
+ ) -> Command {
+ let mut command = self.sync(replica, state, storage);
+ command.arg("--change-set").arg(changes);
+ command
+ }
+
+ pub(super) fn status(&self, replica: &Path, state: &Path) -> Command {
+ let mut command = self.command();
+ command
+ .arg("status")
+ .arg(replica)
+ .arg("--state")
+ .arg(state)
+ .arg("--json");
+ command
+ }
+
+ pub(super) fn gc(&self, storage: &str) -> Command {
+ let mut command = self.command();
+ command.arg("gc").arg(volume_name(storage));
+ command
+ }
+
+ pub(super) fn sync_resolve(
+ &self,
+ replica: &Path,
+ state: &Path,
+ storage: &str,
+ paths: &[&str],
+ ) -> Command {
+ let mut command = self.sync(replica, state, storage);
+ for path in paths {
+ command.arg("--resolve").arg(path);
+ }
+ command
+ }
+
+ fn command(&self) -> Command {
+ let profile = if self.release { "release" } else { "debug" };
+ let mut command = Command::new(
+ PathBuf::from(env!("CARGO_WORKSPACE_DIR"))
+ .join("target")
+ .join(profile)
+ .join("ofs"),
+ );
+ command
+ .env("OFS_HOME", &self.home)
+ .env("AWS_ACCESS_KEY_ID", "minioadmin")
+ .env("AWS_SECRET_ACCESS_KEY", "minioadmin")
+ .env("AWS_REGION", "us-east-1")
+ .env("AWS_EC2_METADATA_DISABLED", "true");
+ if let Some(transfer_concurrency) = self.transfer_concurrency {
+ command.env("OFS_TRANSFER_CONCURRENCY", transfer_concurrency.to_string());
+ }
+ command
+ }
+}
+
+pub(super) fn volume_name(storage: &str) -> String {
+ let root = storage
+ .strip_prefix("s3://managed-sync/")
+ .and_then(|value| value.split('?').next())
+ .unwrap_or(storage);
+ root.replace('/', "-")
+}
+
+pub(super) fn require_success(mut command: Command, action: &str) -> Output {
+ let output = command
+ .output()
+ .unwrap_or_else(|error| panic!("{action}: {error}"));
+ assert!(
+ output.status.success(),
+ "{action} failed: {}",
+ output_text(&output.stderr)
+ );
+ output
+}
+
+pub(super) fn require_failure(mut command: Command, action: &str) -> Output {
+ let output = command
+ .output()
+ .unwrap_or_else(|error| panic!("{action}: {error}"));
+ assert!(!output.status.success(), "{action} unexpectedly succeeded");
+ output
+}
+
+pub(super) fn output_text(bytes: &[u8]) -> String {
+ String::from_utf8_lossy(bytes).trim().to_owned()
+}
+
+pub(super) fn run_logged(command: &mut Command) {
+ println!(
+ "{} {}",
+ command.get_program().to_string_lossy(),
+ command
+ .get_args()
+ .map(|argument| argument.to_string_lossy())
+ .collect::<Vec<_>>()
+ .join(" ")
+ );
+ let status = command.status().expect("failed to execute process");
+ assert!(status.success(), "command failed: {status}");
+}
+
+#[derive(serde::Deserialize)]
+pub(super) struct ManagedStatus {
+ #[serde(skip)]
+ pub(super) document: String,
+ pub(super) volume_id: String,
+ pub(super) common_sequence: u64,
+ pub(super) remote_sequence: u64,
+ pub(super) pending: bool,
+ pub(super) conflicts: u64,
+ pub(super) base_expired: bool,
+ pub(super) capabilities: ManagedCapabilities,
+}
+
+#[derive(serde::Deserialize)]
+pub(super) struct ManagedCapabilities {
+ pub(super) extended_attributes: bool,
+ pub(super) portable_names: bool,
+}
+
+impl ManagedStatus {
+ pub(super) fn parse(document: String) -> Self {
+ let mut status: Self = serde_json::from_str(&document).expect("status is valid JSON");
+ status.document = document;
+ status
+ }
+}
diff --git a/xtask/src/managed_sync/evaluation.rs b/xtask/src/managed_sync/evaluation.rs
new file mode 100644
index 0000000..3aca8bc
--- /dev/null
+++ b/xtask/src/managed_sync/evaluation.rs
@@ -0,0 +1,279 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Explicit network and billing conditions for Managed Sync evaluation.
+
+use clap::Args;
+use std::io::{Read as _, Write as _};
+use std::net::TcpStream;
+use std::sync::mpsc;
+use std::thread;
+use std::time::Duration;
+
+pub(super) const TOXIPROXY_NAME: &str = "managed-sync-minio";
+
+pub(super) struct ChaosController {
+ stop: mpsc::Sender<()>,
+ worker: thread::JoinHandle<ChaosReport>,
+}
+
+#[derive(Clone, Debug)]
+pub(super) struct ChaosReport {
+ pub(super) schedule_seed: String,
+ pub(super) faults_injected: u64,
+}
+
+#[derive(Args, Clone, Debug, Default)]
+pub(crate) struct EvaluationOptions {
+ #[arg(
+ long,
+ default_value_t = 0,
+ help = "Mean object-store round-trip delay in milliseconds."
+ )]
+ pub(super) network_rtt_ms: u64,
+
+ #[arg(
+ long,
+ default_value_t = 0,
+ help = "Random variation around --network-rtt-ms in milliseconds."
+ )]
+ pub(super) network_jitter_ms: u64,
+
+ #[arg(
+ long,
+ default_value_t = 0,
+ help = "Per-connection, per-direction bandwidth limit in decimal KB/s; 0 is unlimited."
+ )]
+ pub(super) connection_bandwidth_kb_per_second: u64,
+
+ #[arg(long, help = "Seed for deterministic connection-reset scheduling.")]
+ pub(super) chaos_seed: Option<u64>,
+
+ #[arg(
+ long,
+ default_value_t = 0,
+ help = "Connection-reset probability per scheduling tick, in millionths."
+ )]
+ pub(super) chaos_reset_rate_per_million: u32,
+
+ #[arg(
+ long,
+ default_value_t = 250,
+ help = "Deterministic chaos scheduling tick in milliseconds."
+ )]
+ pub(super) chaos_tick_ms: u64,
+
+ #[arg(
+ long,
+ default_value_t = 100,
+ help = "Duration of each injected connection outage in milliseconds."
+ )]
+ pub(super) chaos_outage_ms: u64,
+
+ #[arg(
+ long,
+ default_value_t = 0.0,
+ help = "Price per million read requests in USD."
+ )]
+ pub(super) read_request_usd_per_million: f64,
+
+ #[arg(
+ long,
+ default_value_t = 0.0,
+ help = "Price per million write or list requests in USD."
+ )]
+ pub(super) write_list_request_usd_per_million: f64,
+
+ #[arg(
+ long,
+ default_value_t = 0.0,
+ help = "Object-store egress price per GiB in USD."
+ )]
+ pub(super) egress_usd_per_gib: f64,
+
+ #[arg(
+ long,
+ default_value_t = 0.0,
+ help = "Object-store ingress price per GiB in USD."
+ )]
+ pub(super) ingress_usd_per_gib: f64,
+
+ #[arg(
+ long,
+ default_value_t = 0.0,
+ help = "Storage price per GiB-month in USD."
+ )]
+ pub(super) storage_usd_per_gib_month: f64,
+}
+
+impl EvaluationOptions {
+ pub(super) fn validate(&self) {
+ assert!(
+ self.network_jitter_ms <= self.network_rtt_ms,
+ "network jitter must not exceed mean RTT"
+ );
+ assert!(
+ self.connection_bandwidth_kb_per_second <= i64::MAX as u64,
+ "connection bandwidth exceeds the fault proxy range"
+ );
+ assert!(
+ self.chaos_reset_rate_per_million <= 1_000_000,
+ "chaos reset rate must not exceed one million"
+ );
+ assert!(
+ self.chaos_reset_rate_per_million == 0 || self.chaos_seed.is_some(),
+ "--chaos-seed is required when connection resets are enabled"
+ );
+ assert!(
+ self.chaos_reset_rate_per_million == 0 || self.chaos_tick_ms != 0,
+ "chaos tick must be positive when connection resets are enabled"
+ );
+ for (name, value) in [
+ ("read request price", self.read_request_usd_per_million),
+ (
+ "write/list request price",
+ self.write_list_request_usd_per_million,
+ ),
+ ("egress price", self.egress_usd_per_gib),
+ ("ingress price", self.ingress_usd_per_gib),
+ ("storage price", self.storage_usd_per_gib_month),
+ ] {
+ assert!(
+ value.is_finite() && value >= 0.0,
+ "{name} must be finite and non-negative"
+ );
+ }
+ }
+
+ pub(super) const fn network_enabled(&self) -> bool {
+ self.network_rtt_ms != 0
+ || self.network_jitter_ms != 0
+ || self.connection_bandwidth_kb_per_second != 0
+ || self.chaos_reset_rate_per_million != 0
+ }
+
+ pub(super) const fn billing_enabled(&self) -> bool {
+ self.read_request_usd_per_million != 0.0
+ || self.write_list_request_usd_per_million != 0.0
+ || self.egress_usd_per_gib != 0.0
+ || self.ingress_usd_per_gib != 0.0
+ || self.storage_usd_per_gib_month != 0.0
+ }
+
+ pub(super) fn start_chaos(
+ &self,
+ admin_port: u16,
+ root: &str,
+ stage: &str,
+ ) -> Option<ChaosController> {
+ let rate = self.chaos_reset_rate_per_million;
+ if rate == 0 {
+ return None;
+ }
+ let seed = self
+ .chaos_seed
+ .expect("validated chaos configuration has a seed");
+ let mut hasher = blake3::Hasher::new();
+ hasher.update(b"ofs-managed-sync-chaos\0");
+ hasher.update(&seed.to_le_bytes());
+ hasher.update(&(root.len() as u64).to_le_bytes());
+ hasher.update(root.as_bytes());
+ hasher.update(&(stage.len() as u64).to_le_bytes());
+ hasher.update(stage.as_bytes());
+ let schedule_seed = hasher.finalize();
+ let mut schedule = blake3::Hasher::new();
+ schedule.update(b"ofs-managed-sync-chaos-events\0");
+ schedule.update(schedule_seed.as_bytes());
+ let mut schedule = schedule.finalize_xof();
+ let tick = Duration::from_millis(self.chaos_tick_ms);
+ let outage = Duration::from_millis(self.chaos_outage_ms);
+ let reset_threshold = u64::from(rate) * (1_u64 << 32) / 1_000_000;
+ let (stop, receiver) = mpsc::channel();
+ let worker = thread::spawn(move || {
+ let mut faults_injected = 0_u64;
+ loop {
+ match receiver.recv_timeout(tick) {
+ Ok(()) | Err(mpsc::RecvTimeoutError::Disconnected) => break,
+ Err(mpsc::RecvTimeoutError::Timeout) => {}
+ }
+ let mut random = [0_u8; size_of::<u32>()];
+ schedule.fill(&mut random);
+ if u64::from(u32::from_le_bytes(random)) >= reset_threshold {
+ continue;
+ }
+ set_proxy_enabled(admin_port, false)
+ .unwrap_or_else(|error| panic!("inject Managed Sync outage: {error}"));
+ faults_injected += 1;
+ let stopped = match receiver.recv_timeout(outage) {
+ Ok(()) | Err(mpsc::RecvTimeoutError::Disconnected) => true,
+ Err(mpsc::RecvTimeoutError::Timeout) => false,
+ };
+ set_proxy_enabled(admin_port, true)
+ .unwrap_or_else(|error| panic!("end Managed Sync outage: {error}"));
+ if stopped {
+ break;
+ }
+ }
+ ChaosReport {
+ schedule_seed: schedule_seed.to_hex().to_string(),
+ faults_injected,
+ }
+ });
+ Some(ChaosController { stop, worker })
+ }
+}
+
+impl ChaosController {
+ pub(super) fn stop(self) -> ChaosReport {
+ let _ = self.stop.send(());
+ self.worker
+ .join()
+ .expect("Managed Sync chaos controller did not panic")
+ }
+}
+
+fn set_proxy_enabled(port: u16, enabled: bool) -> Result<(), String> {
+ let address = format!("127.0.0.1:{port}");
+ let mut stream = TcpStream::connect_timeout(
+ &address.parse().expect("loopback proxy address is valid"),
+ Duration::from_secs(1),
+ )
+ .map_err(|error| error.to_string())?;
+ stream
+ .set_read_timeout(Some(Duration::from_secs(1)))
+ .map_err(|error| error.to_string())?;
+ stream
+ .set_write_timeout(Some(Duration::from_secs(1)))
+ .map_err(|error| error.to_string())?;
+ let body = format!("{{\"enabled\":{enabled}}}");
+ let request = format!(
+ "POST /proxies/{TOXIPROXY_NAME} HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
+ body.len()
+ );
+ stream
+ .write_all(request.as_bytes())
+ .map_err(|error| error.to_string())?;
+ let mut response = [0; 32];
+ let read = stream
+ .read(&mut response)
+ .map_err(|error| error.to_string())?;
+ if response[..read].starts_with(b"HTTP/1.1 200") {
+ Ok(())
+ } else {
+ Err(String::from_utf8_lossy(&response[..read]).into_owned())
+ }
+}
diff --git a/xtask/src/managed_sync/fixture.rs b/xtask/src/managed_sync/fixture.rs
new file mode 100644
index 0000000..753b1e2
--- /dev/null
+++ b/xtask/src/managed_sync/fixture.rs
@@ -0,0 +1,890 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Disposable MinIO and container fixtures.
+
+use std::collections::BTreeMap;
+use std::env;
+use std::fs;
+use std::io::{Read as _, Write as _};
+use std::net::TcpStream;
+use std::panic::{AssertUnwindSafe, catch_unwind, resume_unwind};
+use std::path::{Path, PathBuf};
+use std::process::Command;
+use std::thread;
+use std::time::{Duration, Instant};
+
+use super::cli::{output_text, run_logged};
+use super::evaluation::{ChaosController, ChaosReport, EvaluationOptions, TOXIPROXY_NAME};
+
+const DEFAULT_MINIO_PORT: u16 = 19_000;
+const DEFAULT_PROXY_PORT: u16 = 19_001;
+const DEFAULT_PROXY_ADMIN_PORT: u16 = 19_002;
+const FIXTURE_READY_TIMEOUT: Duration = Duration::from_secs(30);
+
+pub(super) struct Fixture {
+ compose_file: PathBuf,
+ keep: bool,
+ evaluation: EvaluationOptions,
+ minio_port: u16,
+ ofs_home: PathBuf,
+ _ofs_home_dir: tempfile::TempDir,
+ project: String,
+ proxy_admin_port: u16,
+ proxy_port: u16,
+ started: bool,
+ variant: String,
+}
+
+#[derive(Clone, Copy, Debug, Default)]
+pub(super) struct LogicalIo {
+ pub(super) read_bytes: u64,
+ pub(super) write_bytes: u64,
+}
+
+#[derive(Clone, Debug, Default)]
+struct ProviderSnapshot {
+ requests: BTreeMap<String, u64>,
+ read_bytes: u64,
+ write_bytes: u64,
+}
+
+#[derive(Clone, Debug, Default)]
+struct StorageInventory {
+ classes: BTreeMap<String, StorageClass>,
+}
+
+#[derive(Clone, Copy, Debug, Default)]
+struct StorageClass {
+ objects: u64,
+ bytes: u64,
+}
+
+pub(super) struct CaseRoot {
+ _directory: tempfile::TempDir,
+ pub(super) path: PathBuf,
+}
+
+impl CaseRoot {
+ pub(super) fn new() -> Self {
+ let directory = tempfile::Builder::new()
+ .prefix("opendal-ofs-managed-sync-")
+ .tempdir()
+ .expect("create Managed Sync behavior root");
+ Self {
+ path: directory.path().to_owned(),
+ _directory: directory,
+ }
+ }
+}
+
+#[derive(Debug, Eq, PartialEq)]
+pub(super) struct TreeSummary {
+ pub(super) digest: blake3::Hash,
+ pub(super) files: u64,
+ pub(super) directories: u64,
+ pub(super) bytes: u64,
+}
+
+pub(super) fn tree_summary(root: &Path) -> TreeSummary {
+ let mut fingerprint = blake3::Hasher::new();
+ let mut files = 0_u64;
+ let mut directories = 0_u64;
+ let mut bytes = 0_u64;
+ for entry in walkdir::WalkDir::new(root).min_depth(1).sort_by_file_name() {
+ let entry = entry.expect("read behavior tree");
+ let path = entry.path();
+ let relative = path
+ .strip_prefix(root)
+ .expect("behavior path is below root");
+ let relative = relative.to_str().expect("behavior path is Unicode");
+ let metadata = fs::metadata(path).expect("read behavior metadata");
+ fingerprint.update(&(relative.len() as u64).to_be_bytes());
+ fingerprint.update(relative.as_bytes());
+ if metadata.is_dir() {
+ directories += 1;
+ fingerprint.update(b"d");
+ } else {
+ files += 1;
+ bytes += metadata.len();
+ fingerprint.update(b"f");
+ fingerprint.update(&[u8::from(is_executable(&metadata))]);
+ let mut file = fs::File::open(path).expect("open behavior file");
+ let mut buffer = [0; 1024 * 1024];
+ loop {
+ let read = file.read(&mut buffer).expect("read behavior file");
+ if read == 0 {
+ break;
+ }
+ fingerprint.update(&buffer[..read]);
+ }
+ }
+ }
+ TreeSummary {
+ digest: fingerprint.finalize(),
+ files,
+ directories,
+ bytes,
+ }
+}
+
+#[cfg(unix)]
+fn is_executable(metadata: &fs::Metadata) -> bool {
+ use std::os::unix::fs::PermissionsExt as _;
+
+ metadata.permissions().mode() & 0o111 != 0
+}
+
+#[cfg(not(unix))]
+fn is_executable(_metadata: &fs::Metadata) -> bool {
+ false
+}
+
+impl Fixture {
+ pub(crate) fn new(
+ keep: bool,
+ variant: impl Into<String>,
+ evaluation: EvaluationOptions,
+ ) -> Self {
+ evaluation.validate();
+ let minio_port = configured_port("OFS_MANAGED_SYNC_MINIO_PORT", DEFAULT_MINIO_PORT);
+ let proxy_port = configured_port("OFS_MANAGED_SYNC_PROXY_PORT", DEFAULT_PROXY_PORT);
+ let proxy_admin_port = configured_port(
+ "OFS_MANAGED_SYNC_PROXY_ADMIN_PORT",
+ DEFAULT_PROXY_ADMIN_PORT,
+ );
+ assert!(
+ !evaluation.network_enabled()
+ || (proxy_port != minio_port
+ && proxy_admin_port != minio_port
+ && proxy_admin_port != proxy_port),
+ "Managed Sync fixture ports must differ"
+ );
+ let workspace = PathBuf::from(env!("CARGO_WORKSPACE_DIR"));
+ let ofs_home_dir = tempfile::Builder::new()
+ .prefix("opendal-ofs-home-")
+ .tempdir()
+ .expect("create OFS home");
+ let ofs_home = ofs_home_dir.path().to_owned();
+ Self {
+ compose_file: workspace.join("fixtures/managed-sync/compose.yaml"),
+ keep,
+ evaluation,
+ minio_port,
+ ofs_home,
+ _ofs_home_dir: ofs_home_dir,
+ project: format!("opendal-ofs-managed-sync-{}", std::process::id()),
+ proxy_admin_port,
+ proxy_port,
+ started: false,
+ variant: variant.into(),
+ }
+ }
+
+ pub(super) fn ofs_home(&self) -> &Path {
+ &self.ofs_home
+ }
+
+ pub(super) fn start(mut self) -> Self {
+ self.started = true;
+ if self.evaluation.network_enabled() {
+ run_logged(
+ self.compose()
+ .args(["up", "--detach", "minio", "toxiproxy"]),
+ );
+ } else {
+ run_logged(self.compose().args(["up", "--detach", "minio"]));
+ }
+ self.wait_until_ready(self.minio_port);
+ if self.evaluation.network_enabled() {
+ self.configure_network();
+ self.wait_until_ready(self.proxy_port);
+ }
+ self
+ }
+
+ pub(super) fn start_bub(mut self) -> Self {
+ let workspace = PathBuf::from(env!("CARGO_WORKSPACE_DIR"));
+ run_logged(
+ docker_command()
+ .args(["build", "--network", "host", "--tag", &self.bub_image()])
+ .args(["--file"])
+ .arg(workspace.join("fixtures/managed-sync/bub.Dockerfile"))
+ .arg(&workspace),
+ );
+ self.started = true;
+ run_logged(self.compose().args([
+ "up",
+ "--detach",
+ "--no-build",
+ "minio",
+ "bub-a",
+ "bub-b",
+ ]));
+ self.wait_until_ready(self.minio_port);
+ self
+ }
+
+ fn wait_until_ready(&self, port: u16) {
+ let deadline = Instant::now() + FIXTURE_READY_TIMEOUT;
+ let response_timeout = Duration::from_millis(
+ self.evaluation
+ .network_rtt_ms
+ .saturating_add(self.evaluation.network_jitter_ms)
+ .saturating_add(1_000),
+ );
+ while Instant::now() < deadline {
+ if minio_is_ready(port, response_timeout) {
+ return;
+ }
+ thread::sleep(Duration::from_millis(200));
+ }
+ panic!(
+ "MinIO did not become ready on 127.0.0.1:{} within {} seconds",
+ port,
+ FIXTURE_READY_TIMEOUT.as_secs()
+ );
+ }
+
+ fn configure_network(&self) {
+ let deadline = Instant::now() + FIXTURE_READY_TIMEOUT;
+ loop {
+ let output = self
+ .toxiproxy_cli()
+ .args([
+ "create",
+ "--listen",
+ "0.0.0.0:8666",
+ "--upstream",
+ "minio:9000",
+ TOXIPROXY_NAME,
+ ])
+ .output()
+ .expect("configure Managed Sync fault proxy");
+ if output.status.success() {
+ break;
+ }
+ assert!(
+ Instant::now() < deadline,
+ "configure Managed Sync fault proxy failed: {}",
+ output_text(&output.stderr)
+ );
+ thread::sleep(Duration::from_millis(200));
+ }
+ if self.evaluation.network_rtt_ms != 0 || self.evaluation.network_jitter_ms != 0 {
+ let latency = format!("latency={}", self.evaluation.network_rtt_ms);
+ let jitter = format!("jitter={}", self.evaluation.network_jitter_ms);
+ self.add_toxic([
+ "toxic",
+ "add",
+ "--toxicName",
+ "network-rtt",
+ "--type",
+ "latency",
+ "--attribute",
+ &latency,
+ "--attribute",
+ &jitter,
+ "--downstream",
+ TOXIPROXY_NAME,
+ ]);
+ }
+ let bandwidth = self.evaluation.connection_bandwidth_kb_per_second;
+ if bandwidth != 0 {
+ let rate = format!("rate={bandwidth}");
+ self.add_toxic([
+ "toxic",
+ "add",
+ "--toxicName",
+ "upload-bandwidth",
+ "--type",
+ "bandwidth",
+ "--attribute",
+ &rate,
+ "--upstream",
+ TOXIPROXY_NAME,
+ ]);
+ self.add_toxic([
+ "toxic",
+ "add",
+ "--toxicName",
+ "download-bandwidth",
+ "--type",
+ "bandwidth",
+ "--attribute",
+ &rate,
+ "--downstream",
+ TOXIPROXY_NAME,
+ ]);
+ }
+ }
+
+ fn add_toxic<const N: usize>(&self, arguments: [&str; N]) {
+ let output = self
+ .toxiproxy_cli()
+ .args(arguments)
+ .output()
+ .expect("add Managed Sync network condition");
+ assert!(
+ output.status.success(),
+ "add Managed Sync network condition failed: {}",
+ output_text(&output.stderr)
+ );
+ }
+
+ fn toxiproxy_cli(&self) -> Command {
+ let mut command = self.compose();
+ command.args([
+ "run",
+ "--rm",
+ "--no-deps",
+ "--entrypoint",
+ "/toxiproxy-cli",
+ "toxiproxy",
+ "--host",
+ "http://toxiproxy:8474",
+ ]);
+ command
+ }
+
+ pub(crate) fn create_bucket(&self) {
+ run_logged(self.compose().args([
+ "run",
+ "--rm",
+ "--no-deps",
+ "minio-client",
+ "mb",
+ "--ignore-existing",
+ "local/managed-sync",
+ ]));
+ }
+
+ pub(crate) fn storage_url(&self, root: &str) -> String {
+ format!(
+ "s3://managed-sync/{root}?endpoint=http%3A%2F%2F127.0.0.1%3A{}®ion=us-east-1",
+ self.endpoint_port()
+ )
+ }
+
+ pub(crate) fn storage_usage(&self, target: &str) -> (u64, u64) {
+ let output = self
+ .compose()
+ .args([
+ "run",
+ "--rm",
+ "--no-deps",
+ "-T",
+ "minio-client",
+ "du",
+ "--json",
+ target,
+ ])
+ .output()
+ .expect("inspect Managed storage usage");
+ if !output.status.success() {
+ let missing = serde_json::from_slice::<serde_json::Value>(&output.stdout)
+ .ok()
+ .and_then(|document| document["error"]["message"].as_str().map(str::to_owned))
+ .is_some_and(|message| message.contains("is not a folder"));
+ assert!(
+ missing,
+ "inspect Managed storage usage failed: {}",
+ output_text(&output.stderr)
+ );
+ return (0, 0);
+ }
+ let document: serde_json::Value =
+ serde_json::from_slice(&output.stdout).expect("Managed storage usage is valid JSON");
+ let field = |name| {
+ document[name]
+ .as_u64()
+ .unwrap_or_else(|| panic!("Managed storage usage field {name} is an integer"))
+ };
+ (field("objects"), field("size"))
+ }
+
+ pub(crate) fn observe<T>(
+ &self,
+ root: &str,
+ stage: &str,
+ logical: LogicalIo,
+ action: impl FnOnce() -> T,
+ ) -> T {
+ let before = self.provider_snapshot();
+ let started = Instant::now();
+ let chaos = self
+ .evaluation
+ .start_chaos(self.proxy_admin_port, root, stage);
+ let result = catch_unwind(AssertUnwindSafe(action));
+ let chaos = chaos.map(ChaosController::stop);
+ let elapsed = started.elapsed();
+ let provider = self.provider_snapshot().difference(&before);
+ let inventory = self.storage_inventory(root);
+ report_observation(
+ &self.variant,
+ root,
+ stage,
+ elapsed,
+ logical,
+ &provider,
+ &inventory,
+ &self.evaluation,
+ chaos.as_ref(),
+ );
+ match result {
+ Ok(result) => result,
+ Err(payload) => resume_unwind(payload),
+ }
+ }
+
+ fn provider_snapshot(&self) -> ProviderSnapshot {
+ let output = self
+ .compose()
+ .args([
+ "run",
+ "--rm",
+ "--no-deps",
+ "-T",
+ "minio-client",
+ "admin",
+ "prometheus",
+ "metrics",
+ "--json",
+ "local",
+ "api",
+ "--api-version",
+ "v3",
+ "--bucket",
+ "managed-sync",
+ ])
+ .output()
+ .expect("read MinIO provider metrics");
+ assert!(
+ output.status.success(),
+ "read MinIO provider metrics failed: {}",
+ output_text(&output.stderr)
+ );
+ parse_provider_metrics(&String::from_utf8(output.stdout).expect("MinIO metrics are UTF-8"))
+ }
+
+ fn storage_inventory(&self, root: &str) -> StorageInventory {
+ let target = format!("local/managed-sync/{root}/managed/0/objects");
+ let output = self
+ .compose()
+ .args([
+ "run",
+ "--rm",
+ "--no-deps",
+ "-T",
+ "minio-client",
+ "du",
+ "--recursive",
+ "--depth",
+ "3",
+ "--json",
+ &target,
+ ])
+ .output()
+ .expect("inspect Managed storage inventory");
+ assert!(
+ output.status.success(),
+ "inspect Managed storage inventory failed: {}",
+ output_text(&output.stderr)
+ );
+ let mut inventory = StorageInventory::default();
+ for line in output.stdout.split(|byte| *byte == b'\n') {
+ if line.is_empty() {
+ continue;
+ }
+ let document: serde_json::Value =
+ serde_json::from_slice(line).expect("Managed storage inventory is valid JSON");
+ let prefix = document["prefix"]
+ .as_str()
+ .expect("Managed storage inventory prefix is a string");
+ let Some(relative) =
+ prefix.strip_prefix(&format!("managed-sync/{root}/managed/0/objects/"))
+ else {
+ continue;
+ };
+ let mut parts = relative.split('/');
+ let Some(_epoch) = parts.next() else {
+ continue;
+ };
+ let Some(class) = parts.next() else {
+ continue;
+ };
+ if parts.next().is_some() {
+ continue;
+ }
+ let entry = inventory.classes.entry(class.to_owned()).or_default();
+ entry.objects = entry
+ .objects
+ .checked_add(
+ document["objects"]
+ .as_u64()
+ .expect("Managed storage object count is an integer"),
+ )
+ .expect("Managed storage object count fits u64");
+ entry.bytes = entry
+ .bytes
+ .checked_add(
+ document["size"]
+ .as_u64()
+ .expect("Managed storage object size is an integer"),
+ )
+ .expect("Managed storage byte count fits u64");
+ }
+ inventory
+ }
+
+ pub(super) fn copy_from_container(&self, service: &str, source: &str, destination: &Path) {
+ fs::create_dir(destination).expect("create container evidence directory");
+ let project_label = format!("label=com.docker.compose.project={}", self.project);
+ let service_label = format!("label=com.docker.compose.service={service}");
+ let container = docker_command()
+ .args([
+ "ps",
+ "--filter",
+ &project_label,
+ "--filter",
+ &service_label,
+ "--quiet",
+ ])
+ .output()
+ .expect("resolve fixture container");
+ assert!(
+ container.status.success(),
+ "resolve fixture container failed: {}",
+ output_text(&container.stderr)
+ );
+ let container = output_text(&container.stdout);
+ assert!(
+ !container.is_empty() && !container.contains('\n'),
+ "fixture service did not resolve to exactly one running container"
+ );
+ let remote = format!("{container}:{source}");
+ run_logged(docker_command().args(["cp", &remote]).arg(destination));
+ }
+
+ pub(super) fn compose(&self) -> Command {
+ let mut command = docker_compose();
+ command
+ .env("OFS_BUB_IMAGE", self.bub_image())
+ .env("OFS_MANAGED_SYNC_MINIO_PORT", self.minio_port.to_string())
+ .env(
+ "OFS_MANAGED_SYNC_PROXY_ADMIN_PORT",
+ self.proxy_admin_port.to_string(),
+ )
+ .env("OFS_MANAGED_SYNC_PROXY_PORT", self.proxy_port.to_string())
+ .args(["--project-name", &self.project, "--file"])
+ .arg(&self.compose_file);
+ command
+ }
+
+ fn bub_image(&self) -> String {
+ format!("{}-bub:local", self.project)
+ }
+
+ const fn endpoint_port(&self) -> u16 {
+ if self.evaluation.network_enabled() {
+ self.proxy_port
+ } else {
+ self.minio_port
+ }
+ }
+
+ fn stop(&self) -> bool {
+ self.compose()
+ .args(["down", "--volumes", "--remove-orphans"])
+ .output()
+ .is_ok_and(|output| output.status.success())
+ }
+}
+
+fn configured_port(name: &str, default: u16) -> u16 {
+ env::var(name).map_or(default, |value| {
+ value
+ .parse()
+ .unwrap_or_else(|_| panic!("invalid {name}: {value}"))
+ })
+}
+
+impl Drop for Fixture {
+ fn drop(&mut self) {
+ if !self.started {
+ return;
+ }
+ if self.keep {
+ println!(
+ "Managed Sync fixture retained: project={} port={}",
+ self.project, self.minio_port
+ );
+ } else if !self.stop() {
+ eprintln!("failed to stop Managed Sync fixture {}", self.project);
+ }
+ }
+}
+
+fn docker_compose() -> Command {
+ let mut command = docker_command();
+ command.arg("compose");
+ command
+}
+
+fn docker_command() -> Command {
+ Command::new(
+ which::which("docker")
+ .or_else(|_| which::which("podman"))
+ .unwrap_or_else(|error| panic!("docker or podman not found: {error}")),
+ )
+}
+
+pub(super) fn container_http_proxy() -> Option<String> {
+ ["HTTPS_PROXY", "https_proxy", "HTTP_PROXY", "http_proxy"]
+ .into_iter()
+ .find_map(|name| env::var(name).ok().filter(|value| !value.is_empty()))
+}
+
+fn minio_is_ready(port: u16, response_timeout: Duration) -> bool {
+ let address = format!("127.0.0.1:{port}");
+ let Ok(mut stream) = TcpStream::connect_timeout(
+ &address.parse().expect("loopback fixture address is valid"),
+ Duration::from_secs(1),
+ ) else {
+ return false;
+ };
+ let _ = stream.set_read_timeout(Some(response_timeout));
+ if stream
+ .write_all(
+ b"GET /minio/health/ready HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n",
+ )
+ .is_err()
+ {
+ return false;
+ }
+ let mut response = [0; 32];
+ stream
+ .read(&mut response)
+ .is_ok_and(|read| response[..read].starts_with(b"HTTP/1.1 200"))
+}
+
+impl ProviderSnapshot {
+ fn difference(&self, before: &Self) -> Self {
+ let mut requests = BTreeMap::new();
+ for (name, after) in &self.requests {
+ let delta = after
+ .checked_sub(before.requests.get(name).copied().unwrap_or_default())
+ .unwrap_or_else(|| panic!("MinIO request counter {name} moved backwards"));
+ if delta != 0 {
+ requests.insert(name.clone(), delta);
+ }
+ }
+ Self {
+ requests,
+ read_bytes: self
+ .read_bytes
+ .checked_sub(before.read_bytes)
+ .expect("MinIO sent-byte counter is monotonic"),
+ write_bytes: self
+ .write_bytes
+ .checked_sub(before.write_bytes)
+ .expect("MinIO received-byte counter is monotonic"),
+ }
+ }
+}
+
+fn parse_provider_metrics(document: &str) -> ProviderSnapshot {
+ const REQUESTS: &str = "minio_bucket_api_total";
+ const READ_BYTES: &str = "minio_bucket_api_traffic_received_bytes";
+ const WRITE_BYTES: &str = "minio_bucket_api_traffic_sent_bytes";
+
+ let families: serde_json::Value =
+ serde_json::from_str(document).expect("MinIO provider metrics are valid JSON");
+ let mut snapshot = ProviderSnapshot::default();
+ for family in families
+ .as_array()
+ .expect("MinIO provider metrics are an array")
+ {
+ let family_name = family["name"]
+ .as_str()
+ .expect("MinIO metric family has a name");
+ if !matches!(family_name, REQUESTS | READ_BYTES | WRITE_BYTES) {
+ continue;
+ }
+ for metric in family["metrics"]
+ .as_array()
+ .expect("MinIO metric family has samples")
+ {
+ if metric["labels"]["type"].as_str() != Some("s3") {
+ continue;
+ }
+ let value = parse_counter(&metric["value"])
+ .expect("MinIO counter is a finite unsigned integer");
+ match family_name {
+ REQUESTS => {
+ let request = metric["labels"]["name"]
+ .as_str()
+ .expect("MinIO request metric has a name");
+ *snapshot.requests.entry(request.to_owned()).or_default() += value;
+ }
+ // MinIO reports bucket traffic from the client's direction.
+ READ_BYTES => snapshot.read_bytes += value,
+ WRITE_BYTES => snapshot.write_bytes += value,
+ _ => unreachable!("metric families were filtered"),
+ }
+ }
+ }
+ snapshot
+}
+
+fn parse_counter(value: &serde_json::Value) -> Option<u64> {
+ if let Some(value) = value.as_u64() {
+ return Some(value);
+ }
+ let value = match value {
+ serde_json::Value::String(value) => value.parse::<f64>().ok()?,
+ serde_json::Value::Number(value) => value.as_f64()?,
+ _ => return None,
+ };
+ (value.is_finite() && value >= 0.0 && value.fract() == 0.0 && value <= u64::MAX as f64)
+ .then_some(value as u64)
+}
+
+fn report_observation(
+ variant: &str,
+ root: &str,
+ stage: &str,
+ elapsed: Duration,
+ logical: LogicalIo,
+ provider: &ProviderSnapshot,
+ inventory: &StorageInventory,
+ evaluation: &EvaluationOptions,
+ chaos: Option<&ChaosReport>,
+) {
+ let mut data_objects = 0_u64;
+ let mut data_bytes = 0_u64;
+ let mut metadata_objects = 0_u64;
+ let mut metadata_bytes = 0_u64;
+ let classes = inventory
+ .classes
+ .iter()
+ .map(|(name, class)| {
+ if name == "04-data-segment" {
+ data_objects += class.objects;
+ data_bytes += class.bytes;
+ } else {
+ metadata_objects += class.objects;
+ metadata_bytes += class.bytes;
+ }
+ (
+ name.clone(),
+ serde_json::json!({"objects": class.objects, "bytes": class.bytes}),
+ )
+ })
+ .collect::<serde_json::Map<_, _>>();
+ let request_count = provider.requests.values().sum::<u64>();
+ let amplification = |provider_bytes: u64, logical_bytes: u64| {
+ (logical_bytes != 0).then(|| provider_bytes as f64 / logical_bytes as f64)
+ };
+ let billing = estimate_billing(evaluation, provider, inventory);
+ println!(
+ "managed-sync observation: {}",
+ serde_json::json!({
+ "variant": variant,
+ "scenario": root,
+ "stage": stage,
+ "elapsed_seconds": elapsed.as_secs_f64(),
+ "logical": {
+ "read_bytes": logical.read_bytes,
+ "write_bytes": logical.write_bytes,
+ },
+ "provider": {
+ "requests": provider.requests,
+ "request_count": request_count,
+ "read_bytes": provider.read_bytes,
+ "write_bytes": provider.write_bytes,
+ "read_amplification": amplification(provider.read_bytes, logical.read_bytes),
+ "write_amplification": amplification(provider.write_bytes, logical.write_bytes),
+ },
+ "storage": {
+ "data_objects": data_objects,
+ "data_bytes": data_bytes,
+ "metadata_objects": metadata_objects,
+ "metadata_bytes": metadata_bytes,
+ "classes": classes,
+ },
+ "network": {
+ "rtt_ms": evaluation.network_rtt_ms,
+ "jitter_ms": evaluation.network_jitter_ms,
+ "connection_bandwidth_kb_per_second": evaluation.connection_bandwidth_kb_per_second,
+ },
+ "chaos": chaos.map(|report| serde_json::json!({
+ "seed": evaluation.chaos_seed,
+ "schedule_seed": report.schedule_seed,
+ "reset_rate_per_million": evaluation.chaos_reset_rate_per_million,
+ "tick_ms": evaluation.chaos_tick_ms,
+ "outage_ms": evaluation.chaos_outage_ms,
+ "faults_injected": report.faults_injected,
+ })),
+ "billing": billing,
+ })
+ );
+}
+
+fn estimate_billing(
+ evaluation: &EvaluationOptions,
+ provider: &ProviderSnapshot,
+ inventory: &StorageInventory,
+) -> Option<serde_json::Value> {
+ if !evaluation.billing_enabled() {
+ return None;
+ }
+ let (read_requests, write_list_requests) =
+ provider
+ .requests
+ .iter()
+ .fold((0_u64, 0_u64), |(reads, writes), (name, count)| {
+ if name.starts_with("Get") || name.starts_with("Head") {
+ (reads + count, writes)
+ } else {
+ (reads, writes + count)
+ }
+ });
+ let stored_bytes = inventory
+ .classes
+ .values()
+ .map(|class| class.bytes)
+ .sum::<u64>();
+ let request_usd = read_requests as f64 / 1_000_000.0 * evaluation.read_request_usd_per_million
+ + write_list_requests as f64 / 1_000_000.0 * evaluation.write_list_request_usd_per_million;
+ let gib = 1024.0 * 1024.0 * 1024.0;
+ let egress_usd = provider.read_bytes as f64 / gib * evaluation.egress_usd_per_gib;
+ let ingress_usd = provider.write_bytes as f64 / gib * evaluation.ingress_usd_per_gib;
+ let transfer_usd = egress_usd + ingress_usd;
+ let storage_usd_per_month = stored_bytes as f64 / gib * evaluation.storage_usd_per_gib_month;
+ Some(serde_json::json!({
+ "read_requests": read_requests,
+ "write_list_requests": write_list_requests,
+ "request_usd": request_usd,
+ "egress_usd": egress_usd,
+ "ingress_usd": ingress_usd,
+ "transfer_usd": transfer_usd,
+ "operation_usd": request_usd + transfer_usd,
+ "stored_bytes": stored_bytes,
+ "storage_usd_per_month": storage_usd_per_month,
+ }))
+}
diff --git a/xtask/src/managed_sync/scale.rs b/xtask/src/managed_sync/scale.rs
new file mode 100644
index 0000000..8ee3262
--- /dev/null
+++ b/xtask/src/managed_sync/scale.rs
@@ -0,0 +1,675 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Fixed extreme-scale Managed Sync acceptance.
+
+mod random_write;
+
+use std::fs;
+use std::io::{Seek as _, SeekFrom, Write as _};
+use std::path::{Path, PathBuf};
+use std::time::Instant;
+
+use super::cli::{Ofs, output_text, require_success};
+use super::evaluation::EvaluationOptions;
+use super::fixture::{Fixture, LogicalIo, tree_summary};
+
+const TINY_FILE_COUNT: u64 = 1_000_000;
+const TINY_FILE_BYTES: u64 = 4 * 1024;
+const TINY_CHANGE_COUNT: u64 = 4_096;
+const TINY_DIRECTORY_FANOUT: u64 = 1_000;
+const LARGE_FILE_COUNT: u64 = 3;
+const LARGE_FILE_BYTES: u64 = 10 * 1024 * 1024 * 1024;
+const LARGE_FILE_CHANGED_BYTES: u64 = 3 * 4096;
+const STREAM_BUFFER_BYTES: usize = 1024 * 1024;
+const DATASET_VERSION: &str = "v1";
+
+#[derive(Clone, Copy, Debug, clap::ValueEnum)]
+pub(crate) enum ScaleScenario {
+ TinyFiles,
+ LargeFiles,
+ RandomWrite,
+}
+
+pub(crate) fn run(
+ scenario: ScaleScenario,
+ keep: bool,
+ duration_seconds: u64,
+ sync_interval_seconds: u64,
+ file_size: u64,
+ evaluation: EvaluationOptions,
+) {
+ if matches!(scenario, ScaleScenario::RandomWrite) {
+ random_write::run(
+ keep,
+ duration_seconds,
+ sync_interval_seconds,
+ file_size,
+ evaluation,
+ );
+ return;
+ }
+ let profile = match scenario {
+ ScaleScenario::TinyFiles => Profile::TinyFiles,
+ ScaleScenario::LargeFiles => Profile::LargeFiles,
+ ScaleScenario::RandomWrite => unreachable!("random write was dispatched"),
+ };
+ let fixture = Fixture::new(keep, profile.name(), evaluation);
+ let ofs = Ofs::release(fixture.ofs_home());
+ ofs.build();
+ let fixture = fixture.start();
+ fixture.create_bucket();
+ let work = ScaleRoot::new(keep);
+ let dataset = stage("prepare fixture", || ScaleDataset::open(profile));
+ let replicas = [
+ dataset.path.clone(),
+ work.path.join("replica-b"),
+ work.path.join("replica-c"),
+ ];
+ let states = [
+ work.path.join("state-a"),
+ work.path.join("state-b"),
+ work.path.join("state-c"),
+ ];
+ let changes = [
+ work.path.join("changes-a.ndjson"),
+ work.path.join("changes-b.ndjson"),
+ ];
+ let no_changes = work.path.join("changes-none.ndjson");
+ fs::write(&no_changes, []).expect("create empty scale mutation input");
+ for replica in &replicas[1..] {
+ fs::create_dir(replica).expect("create scale replica");
+ }
+ let storage = fixture.storage_url(&format!("scale/{}", profile.name()));
+
+ remote_stage(
+ &fixture,
+ profile,
+ "create volume",
+ LogicalIo::default(),
+ || {
+ require_success(ofs.volume_create(&storage), "create scale volume");
+ },
+ );
+ remote_stage(
+ &fixture,
+ profile,
+ "initial publish",
+ LogicalIo {
+ read_bytes: 0,
+ write_bytes: profile.logical_bytes(),
+ },
+ || {
+ require_success(
+ ofs.sync(&replicas[0], &states[0], &storage),
+ "publish scale fixture",
+ );
+ },
+ );
+ remote_stage(
+ &fixture,
+ profile,
+ "cold restore",
+ LogicalIo {
+ read_bytes: profile.logical_bytes(),
+ write_bytes: 0,
+ },
+ || {
+ require_success(
+ ofs.sync(&replicas[1], &states[1], &storage),
+ "cold restore scale fixture",
+ );
+ },
+ );
+ require_same_tree(&replicas[0], &replicas[1], "initial cold restore");
+ require_noop(&ofs, &replicas[0], &states[0], &storage, &no_changes);
+ require_noop(&ofs, &replicas[1], &states[1], &storage, &no_changes);
+
+ stage("mutate replica A", || {
+ mutate(
+ profile,
+ Side::A,
+ &replicas[0],
+ &dataset.initial,
+ &changes[0],
+ )
+ });
+ stage("mutate replica B", || {
+ mutate(
+ profile,
+ Side::B,
+ &replicas[1],
+ &dataset.initial,
+ &changes[1],
+ )
+ });
+ remote_stage(
+ &fixture,
+ profile,
+ "publish replica A",
+ LogicalIo {
+ read_bytes: 0,
+ write_bytes: profile.changed_bytes(Side::A),
+ },
+ || {
+ require_success(
+ ofs.sync_changes(&replicas[0], &states[0], &storage, &changes[0]),
+ "publish scale changes from A",
+ );
+ },
+ );
+ remote_stage(
+ &fixture,
+ profile,
+ "merge replica B",
+ LogicalIo {
+ read_bytes: profile.changed_bytes(Side::A),
+ write_bytes: profile.changed_bytes(Side::B),
+ },
+ || {
+ require_success(
+ ofs.sync_changes(&replicas[1], &states[1], &storage, &changes[1]),
+ "merge scale changes from B",
+ );
+ },
+ );
+ remote_stage(
+ &fixture,
+ profile,
+ "converge replica A",
+ LogicalIo {
+ read_bytes: profile.changed_bytes(Side::B),
+ write_bytes: 0,
+ },
+ || {
+ require_success(
+ ofs.sync_changes(&replicas[0], &states[0], &storage, &no_changes),
+ "converge scale replica A",
+ );
+ },
+ );
+ require_same_tree(&replicas[0], &replicas[1], "two-writer convergence");
+
+ remote_stage(
+ &fixture,
+ profile,
+ "collect unreachable data",
+ LogicalIo::default(),
+ || {
+ require_success(ofs.gc(&storage), "collect scale volume");
+ },
+ );
+ remote_stage(
+ &fixture,
+ profile,
+ "post-GC cold restore",
+ LogicalIo {
+ read_bytes: profile.logical_bytes(),
+ write_bytes: 0,
+ },
+ || {
+ require_success(
+ ofs.sync(&replicas[2], &states[2], &storage),
+ "restore scale volume after collection",
+ );
+ },
+ );
+ require_same_tree(&replicas[0], &replicas[2], "post-GC cold restore");
+ for (replica, state) in replicas.iter().zip(&states) {
+ require_noop(&ofs, replica, state, &storage, &no_changes);
+ }
+
+ let summary = tree_summary(&replicas[0]);
+ let state_bytes = states
+ .iter()
+ .map(|path| {
+ fs::metadata(path)
+ .expect("read replica state metadata")
+ .len()
+ })
+ .sum::<u64>();
+ dataset.restore();
+ println!(
+ "Managed Sync scale passed: {} files={} directories={} logical_bytes={} state_bytes={}",
+ profile.name(),
+ summary.files,
+ summary.directories,
+ summary.bytes,
+ state_bytes,
+ );
+}
+
+#[derive(Clone, Copy)]
+enum Profile {
+ TinyFiles,
+ LargeFiles,
+}
+
+impl Profile {
+ const fn name(self) -> &'static str {
+ match self {
+ Self::TinyFiles => "tiny-files",
+ Self::LargeFiles => "large-files",
+ }
+ }
+
+ const fn logical_bytes(self) -> u64 {
+ match self {
+ Self::TinyFiles => TINY_FILE_COUNT * TINY_FILE_BYTES,
+ Self::LargeFiles => LARGE_FILE_COUNT * LARGE_FILE_BYTES,
+ }
+ }
+
+ const fn changed_bytes(self, side: Side) -> u64 {
+ match (self, side) {
+ (Self::TinyFiles, _) => TINY_CHANGE_COUNT / 2 * TINY_FILE_BYTES,
+ (Self::LargeFiles, Side::A) => LARGE_FILE_CHANGED_BYTES,
+ (Self::LargeFiles, Side::B) => (LARGE_FILE_COUNT - 1) * LARGE_FILE_CHANGED_BYTES,
+ }
+ }
+}
+
+#[derive(Clone, Copy)]
+enum Side {
+ A,
+ B,
+}
+
+struct InitialContent(Vec<blake3::Hash>);
+
+impl InitialContent {
+ fn digest(&self, index: u64) -> blake3::Hash {
+ self.0[index as usize]
+ }
+}
+
+fn generate(profile: Profile, root: &Path) -> InitialContent {
+ let mut buffer = vec![0; STREAM_BUFFER_BYTES];
+ match profile {
+ Profile::TinyFiles => {
+ let mut digests = Vec::with_capacity((TINY_CHANGE_COUNT * 2) as usize);
+ for index in 0..TINY_FILE_COUNT {
+ let path = tiny_path(root, index);
+ if index % TINY_DIRECTORY_FANOUT == 0 {
+ fs::create_dir_all(path.parent().expect("tiny file has a parent"))
+ .expect("create tiny-file directory");
+ }
+ let digest = write_xof(&path, index, 0, TINY_FILE_BYTES, &mut buffer);
+ if index < TINY_CHANGE_COUNT * 2 {
+ digests.push(digest);
+ }
+ }
+ InitialContent(digests)
+ }
+ Profile::LargeFiles => {
+ let mut digests = Vec::with_capacity(LARGE_FILE_COUNT as usize);
+ for index in 0..LARGE_FILE_COUNT {
+ digests.push(write_xof(
+ &root.join(format!("large-{index}.bin")),
+ index,
+ 0,
+ LARGE_FILE_BYTES,
+ &mut buffer,
+ ));
+ }
+ InitialContent(digests)
+ }
+ }
+}
+
+fn mutate(profile: Profile, side: Side, root: &Path, initial: &InitialContent, changes: &Path) {
+ let mut output = fs::File::create(changes).expect("create scale mutation input");
+ match profile {
+ Profile::TinyFiles => mutate_tiny(side, root, initial, &mut output),
+ Profile::LargeFiles => mutate_large(side, root, initial, &mut output),
+ }
+}
+
+fn mutate_tiny(side: Side, root: &Path, initial: &InitialContent, changes: &mut fs::File) {
+ let start = match side {
+ Side::A => 0,
+ Side::B => TINY_CHANGE_COUNT,
+ };
+ let quarter = TINY_CHANGE_COUNT / 4;
+ let mut buffer = vec![0; STREAM_BUFFER_BYTES];
+ for index in start..start + quarter {
+ let _ = write_xof(
+ &tiny_path(root, index),
+ index,
+ 1,
+ TINY_FILE_BYTES,
+ &mut buffer,
+ );
+ write_change(
+ changes,
+ &tiny_relative_path(index),
+ initial.digest(index),
+ TINY_FILE_BYTES,
+ &[(0, TINY_FILE_BYTES)],
+ );
+ }
+ for index in start + quarter..start + quarter * 2 {
+ let source = tiny_path(root, index);
+ let destination = source.with_extension("renamed");
+ fs::rename(source, destination).expect("rename tiny scale file");
+ }
+ for index in start + quarter * 2..start + quarter * 3 {
+ fs::remove_file(tiny_path(root, index)).expect("remove tiny scale file");
+ }
+ for index in start + quarter * 3..start + TINY_CHANGE_COUNT {
+ let identity = TINY_FILE_COUNT + index;
+ let path = tiny_path(root, identity);
+ fs::create_dir_all(path.parent().expect("tiny file has a parent"))
+ .expect("create added tiny-file directory");
+ let _ = write_xof(&path, identity, 1, TINY_FILE_BYTES, &mut buffer);
+ }
+}
+
+fn mutate_large(side: Side, root: &Path, initial: &InitialContent, changes: &mut fs::File) {
+ let files: &[u64] = match side {
+ Side::A => &[0],
+ Side::B => &[1, 2],
+ };
+ for index in files {
+ let path = root.join(format!("large-{index}.bin"));
+ let mut file = fs::OpenOptions::new()
+ .write(true)
+ .open(&path)
+ .expect("open large file for partial update");
+ for (offset, marker) in [
+ (0, 0x31_u8),
+ (LARGE_FILE_BYTES / 2, 0x57),
+ (LARGE_FILE_BYTES - 4096, 0x93),
+ ] {
+ file.seek(SeekFrom::Start(offset))
+ .expect("seek large scale file");
+ file.write_all(&[marker; 4096])
+ .expect("write partial large-file update");
+ }
+ file.sync_all().expect("persist large-file update");
+ write_change(
+ changes,
+ &format!("large-{index}.bin"),
+ initial.digest(*index),
+ LARGE_FILE_BYTES,
+ &[
+ (0, 4096),
+ (LARGE_FILE_BYTES / 2, 4096),
+ (LARGE_FILE_BYTES - 4096, 4096),
+ ],
+ );
+ }
+}
+
+fn write_change(
+ output: &mut fs::File,
+ path: &str,
+ digest: blake3::Hash,
+ length: u64,
+ ranges: &[(u64, u64)],
+) {
+ let ranges = ranges
+ .iter()
+ .map(|(offset, length)| serde_json::json!({ "offset": offset, "length": length }))
+ .collect::<Vec<_>>();
+ writeln!(
+ output,
+ "{}",
+ serde_json::json!({
+ "path": path,
+ "base": { "digest": digest.to_hex().to_string(), "length": length },
+ "ranges": ranges,
+ })
+ )
+ .expect("write scale mutation input");
+}
+
+fn tiny_path(root: &Path, index: u64) -> PathBuf {
+ root.join(tiny_relative_path(index))
+}
+
+fn tiny_relative_path(index: u64) -> String {
+ format!("d{:04}/f{index:07}.bin", index / TINY_DIRECTORY_FANOUT)
+}
+
+pub(super) fn write_xof(
+ path: &Path,
+ identity: u64,
+ revision: u8,
+ size: u64,
+ buffer: &mut [u8],
+) -> blake3::Hash {
+ let mut source = content_source(identity, revision);
+ let mut file = fs::File::create(path).expect("create scale file");
+ let mut content = blake3::Hasher::new();
+ let mut remaining = size;
+ while remaining != 0 {
+ let length = remaining.min(buffer.len() as u64) as usize;
+ source.fill(&mut buffer[..length]);
+ file.write_all(&buffer[..length]).expect("write scale file");
+ content.update(&buffer[..length]);
+ remaining -= length as u64;
+ }
+ content.finalize()
+}
+
+pub(super) fn require_same_tree(left: &Path, right: &Path, phase: &str) {
+ let left = tree_summary(left);
+ let right = tree_summary(right);
+ assert_eq!(left, right, "Managed Sync trees differ after {phase}");
+}
+
+fn require_noop(ofs: &Ofs, replica: &Path, state: &Path, storage: &str, changes: &Path) {
+ let output = require_success(
+ ofs.sync_changes(replica, state, storage, changes),
+ "verify scale no-op",
+ );
+ assert!(
+ !output_text(&output.stdout).contains("(published)"),
+ "unchanged scale sync published a new generation"
+ );
+}
+
+fn stage<T>(name: &str, action: impl FnOnce() -> T) -> T {
+ let started = Instant::now();
+ let result = action();
+ println!("scale stage: {name} elapsed={:?}", started.elapsed());
+ result
+}
+
+fn remote_stage(
+ fixture: &Fixture,
+ profile: Profile,
+ name: &str,
+ logical: LogicalIo,
+ action: impl FnOnce(),
+) {
+ fixture.observe(&format!("scale/{}", profile.name()), name, logical, action);
+}
+
+pub(super) struct ScaleRoot {
+ pub(super) path: PathBuf,
+ _directory: tempfile::TempDir,
+}
+
+impl ScaleRoot {
+ pub(super) fn new(keep: bool) -> Self {
+ let scale_root = Path::new(env!("CARGO_WORKSPACE_DIR")).join(".local/scale/runs");
+ fs::create_dir_all(&scale_root).expect("create Managed Sync scale parent");
+ let mut directory = tempfile::Builder::new()
+ .prefix("managed-sync-")
+ .tempdir_in(scale_root)
+ .expect("create Managed Sync scale root");
+ directory.disable_cleanup(keep);
+ if keep {
+ println!(
+ "Managed Sync scale files retained: {}",
+ directory.path().display()
+ );
+ }
+ Self {
+ path: directory.path().to_owned(),
+ _directory: directory,
+ }
+ }
+}
+
+struct ScaleDataset {
+ profile: Profile,
+ path: PathBuf,
+ initial: InitialContent,
+}
+
+impl ScaleDataset {
+ fn open(profile: Profile) -> Self {
+ let root = Path::new(env!("CARGO_WORKSPACE_DIR"))
+ .join(".local/scale/datasets")
+ .join(format!("{}-{DATASET_VERSION}", profile.name()));
+ let path = root.join("data");
+ let manifest = root.join("digests");
+ let initial = match fs::read_to_string(&manifest) {
+ Ok(document) => {
+ println!("scale dataset: reuse {}", path.display());
+ parse_dataset_manifest(profile, &document)
+ }
+ Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
+ if root.exists() {
+ fs::remove_dir_all(&root).expect("remove incomplete scale dataset");
+ }
+ fs::create_dir_all(&path).expect("create scale dataset");
+ println!("scale dataset: generate {}", path.display());
+ let initial = generate(profile, &path);
+ write_dataset_manifest(profile, &root, &initial);
+ return Self {
+ profile,
+ path,
+ initial,
+ };
+ }
+ Err(error) => panic!("read scale dataset manifest: {error}"),
+ };
+ let dataset = Self {
+ profile,
+ path,
+ initial,
+ };
+ dataset.restore();
+ dataset
+ }
+
+ fn restore(&self) {
+ let mut buffer = vec![0; STREAM_BUFFER_BYTES];
+ match self.profile {
+ Profile::TinyFiles => {
+ for index in 0..TINY_CHANGE_COUNT * 2 {
+ let path = tiny_path(&self.path, index);
+ let renamed = path.with_extension("renamed");
+ if let Err(error) = fs::remove_file(&renamed)
+ && error.kind() != std::io::ErrorKind::NotFound
+ {
+ panic!("remove renamed scale file {}: {error}", renamed.display());
+ }
+ let digest = write_xof(&path, index, 0, TINY_FILE_BYTES, &mut buffer);
+ assert_eq!(digest, self.initial.digest(index));
+
+ let added = tiny_path(&self.path, TINY_FILE_COUNT + index);
+ if let Err(error) = fs::remove_file(&added)
+ && error.kind() != std::io::ErrorKind::NotFound
+ {
+ panic!("remove added scale file {}: {error}", added.display());
+ }
+ }
+ }
+ Profile::LargeFiles => {
+ for index in 0..LARGE_FILE_COUNT {
+ let path = self.path.join(format!("large-{index}.bin"));
+ let mut file = fs::OpenOptions::new()
+ .write(true)
+ .open(&path)
+ .expect("open fixed large scale file");
+ assert_eq!(
+ file.metadata().expect("stat fixed large scale file").len(),
+ LARGE_FILE_BYTES,
+ );
+ for offset in [0, LARGE_FILE_BYTES / 2, LARGE_FILE_BYTES - 4096] {
+ let mut source = content_source(index, 0);
+ source.set_position(offset);
+ source.fill(&mut buffer[..4096]);
+ file.seek(SeekFrom::Start(offset))
+ .expect("seek fixed large scale file");
+ file.write_all(&buffer[..4096])
+ .expect("restore fixed large scale file");
+ }
+ file.sync_all().expect("persist fixed large scale file");
+ }
+ }
+ }
+ }
+}
+
+fn parse_dataset_manifest(profile: Profile, document: &str) -> InitialContent {
+ let mut lines = document.lines();
+ assert_eq!(
+ lines.next(),
+ Some(
+ format!(
+ "ofs-managed-sync-scale-{DATASET_VERSION} {}",
+ profile.name()
+ )
+ .as_str()
+ ),
+ "scale dataset manifest does not match this scenario",
+ );
+ let digests = lines
+ .map(|line| line.parse().expect("scale dataset digest is valid BLAKE3"))
+ .collect::<Vec<_>>();
+ let expected = match profile {
+ Profile::TinyFiles => TINY_CHANGE_COUNT * 2,
+ Profile::LargeFiles => LARGE_FILE_COUNT,
+ };
+ assert_eq!(
+ digests.len() as u64,
+ expected,
+ "scale dataset is incomplete"
+ );
+ InitialContent(digests)
+}
+
+fn write_dataset_manifest(profile: Profile, root: &Path, initial: &InitialContent) {
+ let temporary = root.join("digests.tmp");
+ let manifest = root.join("digests");
+ let mut output = fs::File::create(&temporary).expect("create scale dataset manifest");
+ writeln!(
+ output,
+ "ofs-managed-sync-scale-{DATASET_VERSION} {}",
+ profile.name(),
+ )
+ .expect("write scale dataset manifest");
+ for digest in &initial.0 {
+ writeln!(output, "{digest}").expect("write scale dataset digest");
+ }
+ output.sync_all().expect("persist scale dataset manifest");
+ fs::rename(temporary, manifest).expect("publish scale dataset manifest");
+}
+
+fn content_source(identity: u64, revision: u8) -> blake3::OutputReader {
+ let mut seed = blake3::Hasher::new();
+ seed.update(b"ofs-managed-sync-scale\0");
+ seed.update(&identity.to_le_bytes());
+ seed.update(&[revision]);
+ seed.finalize_xof()
+}
diff --git a/xtask/src/managed_sync/scale/random_write.rs b/xtask/src/managed_sync/scale/random_write.rs
new file mode 100644
index 0000000..0ed58b2
--- /dev/null
+++ b/xtask/src/managed_sync/scale/random_write.rs
@@ -0,0 +1,325 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Continuous random writes to one file with periodic Managed Sync runs.
+
+use std::fs;
+use std::io::{Seek as _, SeekFrom, Write as _};
+use std::path::{Path, PathBuf};
+use std::process::Output;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::thread;
+use std::time::{Duration, Instant};
+
+use super::{ScaleRoot, require_same_tree, write_xof};
+use crate::managed_sync::cli::{Ofs, output_text, require_success};
+use crate::managed_sync::evaluation::EvaluationOptions;
+use crate::managed_sync::fixture::{Fixture, LogicalIo};
+
+const WRITE_BYTES: usize = 4 * 1024;
+const STREAM_BUFFER_BYTES: usize = 1024 * 1024;
+const SCENARIO: &str = "scale/random-write";
+
+pub(super) fn run(
+ keep: bool,
+ duration_seconds: u64,
+ sync_interval_seconds: u64,
+ file_size: u64,
+ evaluation: EvaluationOptions,
+) {
+ let workload = Workload::new(duration_seconds, sync_interval_seconds, file_size);
+ let fixture = Fixture::new(keep, "random-write", evaluation);
+ let ofs = Ofs::release(fixture.ofs_home());
+ ofs.build();
+
+ let fixture = fixture.start();
+ fixture.create_bucket();
+ let work = ScaleRoot::new(keep);
+ let source = work.path.join("replica-a");
+ let restored = work.path.join("replica-b");
+ let source_state = work.path.join("state-a");
+ let restored_state = work.path.join("state-b");
+ fs::create_dir(&source).expect("create random-write source replica");
+ fs::create_dir(&restored).expect("create random-write restore replica");
+ let file = source.join("random-write.bin");
+ let storage = fixture.storage_url(SCENARIO);
+
+ let mut buffer = vec![0; STREAM_BUFFER_BYTES];
+ write_xof(&file, 0, 0, workload.file_bytes, &mut buffer);
+ fixture.observe(SCENARIO, "create volume", LogicalIo::default(), || {
+ require_success(ofs.volume_create(&storage), "create random-write volume");
+ });
+ fixture.observe(
+ SCENARIO,
+ "initial publish",
+ LogicalIo {
+ read_bytes: 0,
+ write_bytes: workload.file_bytes,
+ },
+ || {
+ require_success(
+ ofs.sync(&source, &source_state, &storage),
+ "publish random-write fixture",
+ );
+ },
+ );
+
+ let writer = RandomWriter::start(file);
+ let workload_started = Instant::now();
+ let workload_deadline = workload_started
+ .checked_add(workload.duration)
+ .expect("random-write deadline fits Instant");
+ let mut scheduled = workload_started
+ .checked_add(workload.interval)
+ .expect("random-write sync deadline fits Instant");
+ let mut round = 0_u64;
+ let mut previous = WriterSnapshot::default();
+ let mut successful_syncs = 0_u64;
+ let mut failed_syncs = 0_u64;
+ let mut missed_intervals = 0_u64;
+ let mut maximum_sync = Duration::ZERO;
+
+ while scheduled <= workload_deadline {
+ round += 1;
+ wait_until(scheduled);
+ let started = Instant::now();
+ let schedule_lag = started.saturating_duration_since(scheduled);
+ let before = writer.snapshot();
+ let stage = format!("periodic sync {round}");
+ let mut sync_elapsed = Duration::ZERO;
+ let output = fixture.observe(SCENARIO, &stage, LogicalIo::default(), || {
+ let sync_started = Instant::now();
+ let output = run_sync(&ofs, &source, &source_state, &storage);
+ sync_elapsed = sync_started.elapsed();
+ output
+ });
+ let observed_elapsed = started.elapsed();
+ let after = writer.snapshot();
+ let success = output.status.success();
+ successful_syncs += u64::from(success);
+ failed_syncs += u64::from(!success);
+ missed_intervals += u64::from(sync_elapsed > workload.interval);
+ maximum_sync = maximum_sync.max(sync_elapsed);
+ println!(
+ "random-write sync: {}",
+ serde_json::json!({
+ "round": round,
+ "success": success,
+ "scheduled_seconds": scheduled.saturating_duration_since(workload_started).as_secs(),
+ "schedule_lag_seconds": schedule_lag.as_secs_f64(),
+ "sync_elapsed_seconds": sync_elapsed.as_secs_f64(),
+ "observation_elapsed_seconds": observed_elapsed.as_secs_f64(),
+ "within_interval": sync_elapsed <= workload.interval,
+ "writes_before_sync": before.difference(previous),
+ "writes_during_sync": after.difference(before),
+ "error": (!success).then(|| output_text(&output.stderr)),
+ })
+ );
+ previous = after;
+ let Some(next) = scheduled.checked_add(workload.interval) else {
+ break;
+ };
+ scheduled = next;
+ }
+
+ wait_until(workload_deadline);
+ let writer = writer.stop();
+ let tail = writer.difference(previous);
+ let final_output = fixture.observe(SCENARIO, "final sync", LogicalIo::default(), || {
+ run_sync(&ofs, &source, &source_state, &storage)
+ });
+ assert!(
+ final_output.status.success(),
+ "final random-write sync failed: {}",
+ output_text(&final_output.stderr)
+ );
+ fixture.observe(
+ SCENARIO,
+ "cold restore",
+ LogicalIo {
+ read_bytes: workload.file_bytes,
+ write_bytes: 0,
+ },
+ || {
+ require_success(
+ ofs.sync(&restored, &restored_state, &storage),
+ "restore random-write result",
+ );
+ },
+ );
+ require_same_tree(&source, &restored, "random-write cold restore");
+
+ println!(
+ "Managed Sync random-write scale result: {}",
+ serde_json::json!({
+ "duration_seconds": duration_seconds,
+ "sync_interval_seconds": sync_interval_seconds,
+ "file_bytes": workload.file_bytes,
+ "write_bytes": WRITE_BYTES,
+ "writes": {
+ "operations": writer.operations,
+ "bytes": writer.bytes,
+ },
+ "tail_writes": tail,
+ "successful_syncs": successful_syncs,
+ "failed_syncs": failed_syncs,
+ "missed_intervals": missed_intervals,
+ "maximum_sync_seconds": maximum_sync.as_secs_f64(),
+ })
+ );
+ assert_eq!(
+ failed_syncs, 0,
+ "periodic sync failed while the file was being randomly written"
+ );
+ assert_eq!(
+ missed_intervals, 0,
+ "periodic sync did not finish within its scheduling interval"
+ );
+}
+
+fn run_sync(ofs: &Ofs, replica: &Path, state: &Path, storage: &str) -> Output {
+ ofs.sync(replica, state, storage)
+ .output()
+ .expect("run periodic random-write sync")
+}
+
+fn wait_until(deadline: Instant) {
+ while let Some(remaining) = deadline.checked_duration_since(Instant::now()) {
+ thread::sleep(remaining.min(Duration::from_secs(1)));
+ }
+}
+
+struct Workload {
+ file_bytes: u64,
+ duration: Duration,
+ interval: Duration,
+}
+
+impl Workload {
+ fn new(duration_seconds: u64, sync_interval_seconds: u64, file_bytes: u64) -> Self {
+ assert!(
+ duration_seconds != 0,
+ "random-write duration must be positive"
+ );
+ assert!(
+ sync_interval_seconds != 0,
+ "random-write sync interval must be positive"
+ );
+ assert!(
+ file_bytes >= WRITE_BYTES as u64,
+ "random-write file must contain at least one write block"
+ );
+ assert_eq!(
+ file_bytes % WRITE_BYTES as u64,
+ 0,
+ "random-write file size must align to the write block"
+ );
+ Self {
+ file_bytes,
+ duration: Duration::from_secs(duration_seconds),
+ interval: Duration::from_secs(sync_interval_seconds),
+ }
+ }
+}
+
+#[derive(Clone, Copy, Default)]
+struct WriterSnapshot {
+ operations: u64,
+ bytes: u64,
+}
+
+impl WriterSnapshot {
+ fn difference(self, before: Self) -> serde_json::Value {
+ serde_json::json!({
+ "operations": self.operations - before.operations,
+ "bytes": self.bytes - before.bytes,
+ })
+ }
+}
+
+struct RandomWriter {
+ stop: Arc<AtomicBool>,
+ operations: Arc<AtomicU64>,
+ handle: thread::JoinHandle<()>,
+}
+
+impl RandomWriter {
+ fn start(path: PathBuf) -> Self {
+ let stop = Arc::new(AtomicBool::new(false));
+ let operations = Arc::new(AtomicU64::new(0));
+ let thread_stop = Arc::clone(&stop);
+ let thread_operations = Arc::clone(&operations);
+ let handle = thread::spawn(move || {
+ write_randomly(&path, &thread_stop, &thread_operations);
+ });
+ Self {
+ stop,
+ operations,
+ handle,
+ }
+ }
+
+ fn snapshot(&self) -> WriterSnapshot {
+ let operations = self.operations.load(Ordering::Relaxed);
+ WriterSnapshot {
+ operations,
+ bytes: operations * WRITE_BYTES as u64,
+ }
+ }
+
+ fn stop(self) -> WriterSnapshot {
+ let Self {
+ stop,
+ operations,
+ handle,
+ } = self;
+ stop.store(true, Ordering::Relaxed);
+ handle.join().expect("random-write worker did not panic");
+ let operations = operations.load(Ordering::Relaxed);
+ WriterSnapshot {
+ operations,
+ bytes: operations * WRITE_BYTES as u64,
+ }
+ }
+}
+
+fn write_randomly(path: &Path, stop: &AtomicBool, operations: &AtomicU64) {
+ let file_bytes = fs::metadata(path)
+ .expect("read random-write file metadata")
+ .len();
+ let block_count = file_bytes / WRITE_BYTES as u64;
+ let mut file = fs::OpenOptions::new()
+ .write(true)
+ .open(path)
+ .expect("open random-write file");
+ let mut source = blake3::Hasher::new();
+ source.update(b"ofs-managed-sync-random-write\0");
+ let mut source = source.finalize_xof();
+ let mut offset_bytes = [0_u8; 8];
+ let mut payload = [0_u8; WRITE_BYTES];
+ while !stop.load(Ordering::Relaxed) {
+ source.fill(&mut offset_bytes);
+ source.fill(&mut payload);
+ let block = u64::from_le_bytes(offset_bytes) % block_count;
+ file.seek(SeekFrom::Start(block * WRITE_BYTES as u64))
+ .expect("seek random-write file");
+ file.write_all(&payload).expect("update random-write file");
+ operations.fetch_add(1, Ordering::Relaxed);
+ }
+ file.sync_all().expect("persist random-write file");
+}
(opendal-ofs) 02/04: docs: document Managed Sync behavior and scale cargo x commands
Posted by ps...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
PsiACE pushed a commit to branch rewrite/harness
in repository https://gitbox.apache.org/repos/asf/opendal-ofs.git
commit 824b71cef8ccbf8c47bb3d211c7a3843634e3d92
Author: PsiACE <ps...@apache.org>
AuthorDate: Tue Aug 18 02:13:38 2026 +0800
docs: document Managed Sync behavior and scale cargo x commands
---
CONTRIBUTING.md | 3 +++
1 file changed, 3 insertions(+)
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 7999f96..a9d1d5f 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -29,6 +29,9 @@ and CI. Run `cargo x --help` to list its commands.
- `cargo x lint` checks Rust formatting, Clippy, documentation, TOML formatting,
and spelling.
- `cargo x licenses` checks source headers and dependency licenses.
+- `cargo x behavior managed-sync` runs the black-box Managed Sync behavior suite.
+- `cargo x scale managed-sync` and `cargo x e2e managed-sync-bub` are scale and
+ Bub acceptance; they are not ordinary PR gates.
Use `cargo x lint --fix` to apply the available formatting and Clippy fixes.
(opendal-ofs) 04/04: test(harness): drive random-write scale through FileChangeSet
Posted by ps...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
PsiACE pushed a commit to branch rewrite/harness
in repository https://gitbox.apache.org/repos/asf/opendal-ofs.git
commit 62e72803b15791e4822d436b22027255154a380a
Author: PsiACE <ps...@apache.org>
AuthorDate: Tue Aug 18 10:19:53 2026 +0800
test(harness): drive random-write scale through FileChangeSet
The writer keeps running. Each periodic sync takes the dirty ranges
accumulated since the last attempt and passes them as --change-set.
---
xtask/src/managed_sync/scale/random_write.rs | 176 +++++++++++++++++++++++++--
1 file changed, 165 insertions(+), 11 deletions(-)
diff --git a/xtask/src/managed_sync/scale/random_write.rs b/xtask/src/managed_sync/scale/random_write.rs
index 0ed58b2..78751d6 100644
--- a/xtask/src/managed_sync/scale/random_write.rs
+++ b/xtask/src/managed_sync/scale/random_write.rs
@@ -21,8 +21,8 @@ use std::fs;
use std::io::{Seek as _, SeekFrom, Write as _};
use std::path::{Path, PathBuf};
use std::process::Output;
-use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};
@@ -60,7 +60,7 @@ pub(super) fn run(
let storage = fixture.storage_url(SCENARIO);
let mut buffer = vec![0; STREAM_BUFFER_BYTES];
- write_xof(&file, 0, 0, workload.file_bytes, &mut buffer);
+ let base_digest = write_xof(&file, 0, 0, workload.file_bytes, &mut buffer);
fixture.observe(SCENARIO, "create volume", LogicalIo::default(), || {
require_success(ofs.volume_create(&storage), "create random-write volume");
});
@@ -79,7 +79,8 @@ pub(super) fn run(
},
);
- let writer = RandomWriter::start(file);
+ let changes = work.path.join("changes.ndjson");
+ let writer = RandomWriter::start(file.clone(), workload.file_bytes);
let workload_started = Instant::now();
let workload_deadline = workload_started
.checked_add(workload.duration)
@@ -104,8 +105,20 @@ pub(super) fn run(
let mut sync_elapsed = Duration::ZERO;
let output = fixture.observe(SCENARIO, &stage, LogicalIo::default(), || {
let sync_started = Instant::now();
- let output = run_sync(&ofs, &source, &source_state, &storage);
+ let (output, rejected) = run_change_set_sync(
+ &ofs,
+ &source,
+ &source_state,
+ &storage,
+ &changes,
+ &writer,
+ base_digest,
+ workload.file_bytes,
+ );
sync_elapsed = sync_started.elapsed();
+ if let Some(dirty) = rejected {
+ writer.restore_dirty(&dirty);
+ }
output
});
let observed_elapsed = started.elapsed();
@@ -141,7 +154,9 @@ pub(super) fn run(
let writer = writer.stop();
let tail = writer.difference(previous);
let final_output = fixture.observe(SCENARIO, "final sync", LogicalIo::default(), || {
- run_sync(&ofs, &source, &source_state, &storage)
+ ofs.sync(&source, &source_state, &storage)
+ .output()
+ .expect("run final random-write sync")
});
assert!(
final_output.status.success(),
@@ -192,10 +207,56 @@ pub(super) fn run(
);
}
-fn run_sync(ofs: &Ofs, replica: &Path, state: &Path, storage: &str) -> Output {
- ofs.sync(replica, state, storage)
+fn run_change_set_sync(
+ ofs: &Ofs,
+ replica: &Path,
+ state: &Path,
+ storage: &str,
+ changes: &Path,
+ writer: &RandomWriter,
+ base_digest: blake3::Hash,
+ file_bytes: u64,
+) -> (Output, Option<Vec<(u64, u64)>>) {
+ let dirty = writer.take_dirty();
+ write_change_set(changes, "random-write.bin", base_digest, file_bytes, &dirty);
+ let output = ofs
+ .sync_changes(replica, state, storage, changes)
.output()
- .expect("run periodic random-write sync")
+ .expect("run random-write change-set sync");
+ if output.status.success() {
+ (output, None)
+ } else {
+ (output, Some(dirty))
+ }
+}
+
+fn write_change_set(
+ output: &Path,
+ path: &str,
+ digest: blake3::Hash,
+ length: u64,
+ ranges: &[(u64, u64)],
+) {
+ if ranges.is_empty() {
+ fs::write(output, []).expect("write empty random-write change set");
+ return;
+ }
+ let ranges = ranges
+ .iter()
+ .map(|(offset, length)| serde_json::json!({ "offset": offset, "length": length }))
+ .collect::<Vec<_>>();
+ fs::write(
+ output,
+ format!(
+ "{}\n",
+ serde_json::json!({
+ "path": path,
+ "base": { "digest": digest.to_hex().to_string(), "length": length },
+ "ranges": ranges,
+ })
+ ),
+ )
+ .expect("write random-write change set");
}
fn wait_until(deadline: Instant) {
@@ -254,26 +315,46 @@ impl WriterSnapshot {
struct RandomWriter {
stop: Arc<AtomicBool>,
+ dirty: Arc<Mutex<DirtyBlocks>>,
operations: Arc<AtomicU64>,
handle: thread::JoinHandle<()>,
}
impl RandomWriter {
- fn start(path: PathBuf) -> Self {
+ fn start(path: PathBuf, file_bytes: u64) -> Self {
let stop = Arc::new(AtomicBool::new(false));
+ let dirty = Arc::new(Mutex::new(DirtyBlocks::new(
+ file_bytes / WRITE_BYTES as u64,
+ )));
let operations = Arc::new(AtomicU64::new(0));
let thread_stop = Arc::clone(&stop);
+ let thread_dirty = Arc::clone(&dirty);
let thread_operations = Arc::clone(&operations);
let handle = thread::spawn(move || {
- write_randomly(&path, &thread_stop, &thread_operations);
+ write_randomly(&path, &thread_stop, &thread_dirty, &thread_operations);
});
Self {
stop,
+ dirty,
operations,
handle,
}
}
+ fn take_dirty(&self) -> Vec<(u64, u64)> {
+ self.dirty
+ .lock()
+ .expect("take random-write dirty ranges")
+ .take()
+ }
+
+ fn restore_dirty(&self, ranges: &[(u64, u64)]) {
+ self.dirty
+ .lock()
+ .expect("restore random-write dirty ranges")
+ .mark_ranges(ranges);
+ }
+
fn snapshot(&self) -> WriterSnapshot {
let operations = self.operations.load(Ordering::Relaxed);
WriterSnapshot {
@@ -287,6 +368,7 @@ impl RandomWriter {
stop,
operations,
handle,
+ dirty: _,
} = self;
stop.store(true, Ordering::Relaxed);
handle.join().expect("random-write worker did not panic");
@@ -298,7 +380,12 @@ impl RandomWriter {
}
}
-fn write_randomly(path: &Path, stop: &AtomicBool, operations: &AtomicU64) {
+fn write_randomly(
+ path: &Path,
+ stop: &AtomicBool,
+ dirty: &Mutex<DirtyBlocks>,
+ operations: &AtomicU64,
+) {
let file_bytes = fs::metadata(path)
.expect("read random-write file metadata")
.len();
@@ -319,7 +406,74 @@ fn write_randomly(path: &Path, stop: &AtomicBool, operations: &AtomicU64) {
file.seek(SeekFrom::Start(block * WRITE_BYTES as u64))
.expect("seek random-write file");
file.write_all(&payload).expect("update random-write file");
+ dirty
+ .lock()
+ .expect("record random-write dirty block")
+ .mark(block);
operations.fetch_add(1, Ordering::Relaxed);
}
file.sync_all().expect("persist random-write file");
}
+
+struct DirtyBlocks {
+ bits: Vec<u64>,
+ block_count: u64,
+}
+
+impl DirtyBlocks {
+ fn new(block_count: u64) -> Self {
+ let words = usize::try_from(block_count.div_ceil(64)).expect("dirty bitmap fits");
+ Self {
+ bits: vec![0; words],
+ block_count,
+ }
+ }
+
+ fn mark(&mut self, block: u64) {
+ let index = usize::try_from(block).expect("dirty block fits");
+ self.bits[index / 64] |= 1_u64 << (index % 64);
+ }
+
+ fn mark_ranges(&mut self, ranges: &[(u64, u64)]) {
+ for &(offset, length) in ranges {
+ let start = offset / WRITE_BYTES as u64;
+ let end = (offset + length).div_ceil(WRITE_BYTES as u64);
+ for block in start..end.min(self.block_count) {
+ self.mark(block);
+ }
+ }
+ }
+
+ fn take(&mut self) -> Vec<(u64, u64)> {
+ let ranges = self.ranges();
+ self.bits.fill(0);
+ ranges
+ }
+
+ fn ranges(&self) -> Vec<(u64, u64)> {
+ let mut ranges = Vec::new();
+ let mut start = None;
+ for block in 0..self.block_count {
+ let index = usize::try_from(block).expect("dirty block fits");
+ let dirty = self.bits[index / 64] & (1_u64 << (index % 64)) != 0;
+ match (dirty, start) {
+ (true, None) => start = Some(block),
+ (false, Some(first)) => {
+ ranges.push((
+ first * WRITE_BYTES as u64,
+ (block - first) * WRITE_BYTES as u64,
+ ));
+ start = None;
+ }
+ _ => {}
+ }
+ }
+ if let Some(first) = start {
+ ranges.push((
+ first * WRITE_BYTES as u64,
+ (self.block_count - first) * WRITE_BYTES as u64,
+ ));
+ }
+ ranges
+ }
+}
(opendal-ofs) 03/04: ci: require Managed Sync behavior on pull requests
Posted by ps...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
PsiACE pushed a commit to branch rewrite/harness
in repository https://gitbox.apache.org/repos/asf/opendal-ofs.git
commit c511fda8e31e00beafe7709005c9973f9789a53c
Author: PsiACE <ps...@apache.org>
AuthorDate: Tue Aug 18 02:13:38 2026 +0800
ci: require Managed Sync behavior on pull requests
---
.github/workflows/ci.yml | 18 ++++++++++++++++++
1 file changed, 18 insertions(+)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 6c11456..5490b0c 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -107,12 +107,29 @@ jobs:
- name: Check licenses
run: cargo x licenses
+ managed-behavior:
+ name: Managed behavior
+ runs-on: ubuntu-24.04
+ timeout-minutes: 15
+ steps:
+ - uses: actions/checkout@v7
+
+ - name: Show fixture tools
+ run: |
+ rustc --version
+ cargo --version
+ docker compose version
+
+ - name: Run Managed Sync behavior
+ run: cargo x behavior managed-sync
+
required:
name: Required
runs-on: ubuntu-24.04
if: ${{ always() }}
needs:
- check
+ - managed-behavior
- msrv
- test
- compliance
@@ -121,6 +138,7 @@ jobs:
run: |
if [[ ! ( \
"${{ needs.check.result }}" == "success" \
+ && "${{ needs.managed-behavior.result }}" == "success" \
&& "${{ needs.msrv.result }}" == "success" \
&& "${{ needs.test.result }}" == "success" \
&& "${{ needs.compliance.result }}" == "success" \