You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2026/10/26 17:33:56 UTC
(opendal-yinyang) branch xuanwo/managed-sync-cli-probe updated: fix: compile and harden the Managed Sync CLI adapter
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch xuanwo/managed-sync-cli-probe
in repository https://gitbox.apache.org/repos/asf/opendal-yinyang.git
The following commit(s) were added to refs/heads/xuanwo/managed-sync-cli-probe by this push:
new c41f369 fix: compile and harden the Managed Sync CLI adapter
c41f369 is described below
commit c41f369d2e1d104766748b381f55fde2464625d5
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Oct 27 01:33:50 2026 +0800
fix: compile and harden the Managed Sync CLI adapter
Resolve the CLI module against the library's `yinyang::volume` path,
require Managed + Sync admission explicitly, propagate status
serialization and stdout failure, and add CLI-level admission and
exit-code tests. Trim the status projection to a Serialize-only DTO set.
Co-authored-by: Chojan Shang <ps...@apache.org>
---
src/main.rs | 4 +-
src/sync_cli.rs | 111 ++++++++++++++++++++++++-------------------
tests/sync_cli.rs | 140 ++++++++++++++++++++++++++++++++++++++++++++++++++++++
3 files changed, 204 insertions(+), 51 deletions(-)
diff --git a/src/main.rs b/src/main.rs
index 378ad6b..9a04c14 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -130,7 +130,7 @@ async fn run(cli: Cli) -> Result<ExitCode, Box<dyn std::error::Error>> {
let listener = tokio::net::TcpListener::bind(listen).await?;
println!("metadata service listening on {}", listener.local_addr()?);
yinyang::core::service::serve(listener, service, token).await?;
- return Ok(());
+ return Ok(ExitCode::SUCCESS);
}
if matches!(cli.command, Command::Create) {
let fs = Fs::create(operator, profile).await?;
@@ -138,7 +138,7 @@ async fn run(cli: Cli) -> Result<ExitCode, Box<dyn std::error::Error>> {
"filesystem ready at revision {}",
revision(fs.observe_latest().await?.revision())
);
- return Ok(());
+ return Ok(ExitCode::SUCCESS);
}
let fs = Fs::open(operator, profile).await?;
let observed = fs.observe_latest().await?;
diff --git a/src/sync_cli.rs b/src/sync_cli.rs
index 0fb920c..cade7a3 100644
--- a/src/sync_cli.rs
+++ b/src/sync_cli.rs
@@ -23,11 +23,11 @@
//! and `sync`); this adapter binds a volume configuration to that contract and
//! formats its results.
-use crate::volume::VolumeConfig;
use clap::{Subcommand, ValueEnum};
-use serde::{Deserialize, Serialize};
+use serde::Serialize;
use std::path::{Path, PathBuf};
use std::process::ExitCode;
+use yinyang::volume::{Access, Model, VolumeConfig};
/// Managed Sync subcommands. Every subcommand requires `--volume CONFIG`.
#[derive(Subcommand)]
@@ -97,7 +97,7 @@ const EXIT_INVALID_BINDING: u8 = 4;
///
/// Every field is always present; unknown values are reported as `null` rather
/// than omitted, so a consumer can distinguish "unknown" from "not applicable".
-#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Debug, Serialize, Eq, PartialEq)]
pub struct SyncStatus {
pub schema: String,
pub binding: BindingStatus,
@@ -140,7 +140,7 @@ impl SyncStatus {
}
}
-#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Debug, Serialize, Eq, PartialEq)]
pub struct BindingStatus {
pub state: BindingState,
pub state_directory: String,
@@ -149,7 +149,7 @@ pub struct BindingStatus {
pub same_filesystem: Option<bool>,
}
-#[derive(Clone, Copy, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Copy, Debug, Serialize, Eq, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum BindingState {
Unbound,
@@ -157,14 +157,14 @@ pub enum BindingState {
Error,
}
-#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Debug, Serialize, Eq, PartialEq)]
pub struct RemoteStatus {
pub availability: RemoteAvailability,
pub revision: Option<String>,
pub error: Option<String>,
}
-#[derive(Clone, Copy, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Copy, Debug, Serialize, Eq, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum RemoteAvailability {
NotChecked,
@@ -172,13 +172,13 @@ pub enum RemoteAvailability {
Unavailable,
}
-#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Debug, Serialize, Eq, PartialEq)]
pub struct LocalStatus {
pub scan: LocalScanState,
pub error: Option<String>,
}
-#[derive(Clone, Copy, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Copy, Debug, Serialize, Eq, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum LocalScanState {
NotScanned,
@@ -187,7 +187,7 @@ pub enum LocalScanState {
Failed,
}
-#[derive(Clone, Debug, Default, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Debug, Default, Serialize, Eq, PartialEq)]
pub struct StatusSummary {
pub in_sync: bool,
pub pending: u64,
@@ -197,7 +197,7 @@ pub struct StatusSummary {
pub retained_artifacts: u64,
}
-#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Debug, Serialize, Eq, PartialEq)]
pub struct SyncEntry {
pub path: String,
pub original_path: Option<String>,
@@ -211,7 +211,7 @@ pub struct SyncEntry {
pub conflict: Option<EntryConflict>,
}
-#[derive(Clone, Copy, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Copy, Debug, Serialize, Eq, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum EntryState {
InSync,
@@ -222,7 +222,7 @@ pub enum EntryState {
Failed,
}
-#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Debug, Serialize, Eq, PartialEq)]
pub struct LocalEvidence {
pub kind: Option<EntryKind>,
pub length: Option<u64>,
@@ -231,7 +231,7 @@ pub struct LocalEvidence {
pub presence: Presence,
}
-#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Debug, Serialize, Eq, PartialEq)]
pub struct RemoteEvidence {
pub kind: Option<EntryKind>,
pub length: Option<u64>,
@@ -242,14 +242,14 @@ pub struct RemoteEvidence {
pub revision: Option<String>,
}
-#[derive(Clone, Copy, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Copy, Debug, Serialize, Eq, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum EntryKind {
File,
Directory,
}
-#[derive(Clone, Copy, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Copy, Debug, Serialize, Eq, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum Presence {
Present,
@@ -257,13 +257,13 @@ pub enum Presence {
Unknown,
}
-#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Debug, Serialize, Eq, PartialEq)]
pub struct BaselineEvidence {
pub node: Option<String>,
pub revision: Option<String>,
}
-#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Debug, Serialize, Eq, PartialEq)]
pub struct IntentStatus {
pub id: Option<String>,
pub operation: Option<IntentOperation>,
@@ -272,7 +272,7 @@ pub struct IntentStatus {
pub receipt_revision: Option<String>,
}
-#[derive(Clone, Copy, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Copy, Debug, Serialize, Eq, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum IntentOperation {
Content,
@@ -283,7 +283,7 @@ pub enum IntentOperation {
SetExecutable,
}
-#[derive(Clone, Copy, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Copy, Debug, Serialize, Eq, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum IntentState {
Prepared,
@@ -294,13 +294,13 @@ pub enum IntentState {
Failed,
}
-#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Debug, Serialize, Eq, PartialEq)]
pub struct OutcomeStatus {
pub state: OutcomeState,
pub error: Option<String>,
}
-#[derive(Clone, Copy, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Copy, Debug, Serialize, Eq, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum OutcomeState {
Pending,
@@ -312,7 +312,7 @@ pub enum OutcomeState {
None,
}
-#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Debug, Serialize, Eq, PartialEq)]
pub struct EntryConflict {
pub id: String,
pub local: ConflictSide,
@@ -322,7 +322,7 @@ pub struct EntryConflict {
pub remote_revision: Option<String>,
}
-#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Debug, Serialize, Eq, PartialEq)]
pub struct ConflictRecord {
pub id: String,
pub path: String,
@@ -336,7 +336,7 @@ pub struct ConflictRecord {
pub error: String,
}
-#[derive(Clone, Copy, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Copy, Debug, Serialize, Eq, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum ConflictSide {
CurrentFile,
@@ -346,7 +346,7 @@ pub enum ConflictSide {
Unknown,
}
-#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Debug, Serialize, Eq, PartialEq)]
pub struct ArtifactRecord {
pub id: String,
pub origin: String,
@@ -357,7 +357,7 @@ pub struct ArtifactRecord {
pub error: Option<String>,
}
-#[derive(Clone, Copy, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Copy, Debug, Serialize, Eq, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum ArtifactKind {
ReplacedFile,
@@ -367,7 +367,7 @@ pub enum ArtifactKind {
UploadRetained,
}
-#[derive(Clone, Copy, Debug, Serialize, Deserialize, Eq, PartialEq)]
+#[derive(Clone, Copy, Debug, Serialize, Eq, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum ArtifactPhase {
Preparing,
@@ -428,6 +428,15 @@ fn operator() -> Result<opendal::Operator, opendal::Error> {
/// Applies the same four-layer admission as every other volume command.
fn admit(config: &VolumeConfig) -> Result<(), Box<dyn std::error::Error>> {
+ // `VolumeConfig::validate` admits every supported mode; the sync frontend
+ // must additionally pin the exact Managed + Sync combination it implements.
+ if config.model != Model::Managed || config.access != Access::Sync {
+ return Err(format!(
+ "managed sync requires a Managed volume with access=sync, found model={:?} access={:?}",
+ config.model, config.access
+ )
+ .into());
+ }
let operator = operator().map_err(|error| format!("storage configuration: {error}"))?;
config.validate(&operator)?;
Ok(())
@@ -438,11 +447,11 @@ async fn status(config: &VolumeConfig, json: bool) -> Result<ExitCode, Box<dyn s
// stays offline and never needs the authority.
if !config.staging.is_dir() {
let document = SyncStatus::unbound(&config.staging);
- emit(&document, json);
+ emit(&document, json)?;
return Ok(ExitCode::from(EXIT_UNBOUND));
}
let inspection = manager::inspect(config).await?;
- emit(&inspection.status, json);
+ emit(&inspection.status, json)?;
Ok(ExitCode::from(inspection.exit))
}
@@ -503,26 +512,30 @@ async fn resolve(
Ok(ExitCode::SUCCESS)
}
-fn emit(document: &SyncStatus, json: bool) {
+/// Writes the status projection, propagating serialization or stdout failure
+/// instead of reporting success without a document.
+fn emit(document: &SyncStatus, json: bool) -> Result<(), Box<dyn std::error::Error>> {
+ use std::io::Write as _;
+ let mut stdout = std::io::stdout().lock();
if json {
- match serde_json::to_string_pretty(document) {
- Ok(text) => println!("{text}"),
- Err(error) => eprintln!("yy: serialize sync status: {error}"),
- }
- return;
+ let text = serde_json::to_string_pretty(document)?;
+ writeln!(stdout, "{text}")?;
+ } else {
+ writeln!(
+ stdout,
+ "sync: {:?} remote: {:?} local: {:?} in-sync: {} pending: {} unknown: {} conflicts: {} failed: {} artifacts: {}",
+ document.binding.state,
+ document.remote.availability,
+ document.local.scan,
+ document.summary.in_sync,
+ document.summary.pending,
+ document.summary.unknown,
+ document.summary.conflicts,
+ document.summary.failed,
+ document.summary.retained_artifacts,
+ )?;
}
- println!(
- "sync: {:?} remote: {:?} local: {:?} in-sync: {} pending: {} unknown: {} conflicts: {} failed: {} artifacts: {}",
- document.binding.state,
- document.remote.availability,
- document.local.scan,
- document.summary.in_sync,
- document.summary.pending,
- document.summary.unknown,
- document.summary.conflicts,
- document.summary.failed,
- document.summary.retained_artifacts,
- );
+ Ok(())
}
/// Boundary between the CLI adapter and the library's reconciliation contract.
@@ -533,8 +546,8 @@ fn emit(document: &SyncStatus, json: bool) {
/// single place that has to be wired to that public API.
mod manager {
use super::SyncStatus;
- use crate::volume::VolumeConfig;
use std::path::Path;
+ use yinyang::volume::VolumeConfig;
pub struct Inspection {
pub status: SyncStatus,
diff --git a/tests/sync_cli.rs b/tests/sync_cli.rs
new file mode 100644
index 0000000..83b1db3
--- /dev/null
+++ b/tests/sync_cli.rs
@@ -0,0 +1,140 @@
+// 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.
+
+//! CLI admission and error behavior for `yy ... sync`.
+//!
+//! The exact status schema is exercised end to end by the Managed Sync
+//! acceptance suite; these tests only pin the frontend's admission rules and
+//! exit-code contract so they fail fast without storage.
+
+#![cfg(target_os = "linux")]
+
+use std::path::{Path, PathBuf};
+use std::process::{Command, Output};
+
+fn yy(args: &[&str]) -> Output {
+ Command::new(env!("CARGO_BIN_EXE_yy"))
+ .args(args)
+ // Dummy S3 settings are enough for admission: capability negotiation
+ // reads declared operator capabilities and never contacts storage.
+ .env("YINYANG_S3_BUCKET", "cli-admission")
+ .env("YINYANG_S3_REGION", "us-east-1")
+ .env("YINYANG_S3_ENDPOINT", "http://127.0.0.1:9")
+ .env("YINYANG_STORAGE_PROFILE", "minio")
+ .output()
+ .expect("run yy")
+}
+
+fn write_config(dir: &Path, name: &str, model: &str, access: &str) -> PathBuf {
+ let staging = dir.join(format!("{name}-state"));
+ let config = dir.join(format!("{name}.json"));
+ let value = serde_json::json!({
+ "name": name,
+ "model": model,
+ "access": access,
+ "frontend": "library",
+ "publication": "object",
+ "storage_profile": "minio",
+ "staging": staging,
+ "read_only": false,
+ "require": [],
+ });
+ std::fs::write(&config, serde_json::to_vec_pretty(&value).unwrap()).unwrap();
+ config
+}
+
+fn assert_stderr(output: &Output, needle: &str) {
+ let stderr = String::from_utf8_lossy(&output.stderr);
+ assert!(
+ stderr.contains(needle),
+ "stderr did not contain {needle:?}: {stderr}"
+ );
+}
+
+#[test]
+fn sync_requires_a_volume() {
+ let output = yy(&["sync", "status"]);
+ assert!(!output.status.success());
+ assert_stderr(&output, "requires --volume CONFIG.json");
+}
+
+#[test]
+fn sync_rejects_a_mount_volume() {
+ let temp = tempfile::tempdir().unwrap();
+ let config = write_config(temp.path(), "mount", "managed", "mount");
+ let output = yy(&["--volume", config.to_str().unwrap(), "sync", "status"]);
+ assert!(!output.status.success());
+ assert_stderr(&output, "requires a Managed volume with access=sync");
+}
+
+#[test]
+fn sync_rejects_a_direct_volume() {
+ let temp = tempfile::tempdir().unwrap();
+ let config = write_config(temp.path(), "direct", "direct", "sync");
+ let output = yy(&["--volume", config.to_str().unwrap(), "sync", "status"]);
+ assert!(!output.status.success());
+ assert_stderr(&output, "requires a Managed volume with access=sync");
+}
+
+#[test]
+fn unbound_status_is_schema_valid_and_exits_three() {
+ let temp = tempfile::tempdir().unwrap();
+ let config = write_config(temp.path(), "unbound", "managed", "sync");
+ let output = yy(&[
+ "--volume",
+ config.to_str().unwrap(),
+ "sync",
+ "status",
+ "--json",
+ ]);
+ assert_eq!(output.status.code(), Some(3));
+ let value: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap();
+ assert_eq!(value["schema"], "yinyang.sync.status.v1");
+ assert_eq!(value["binding"]["state"], "unbound");
+}
+
+#[test]
+fn export_requires_a_selector() {
+ let temp = tempfile::tempdir().unwrap();
+ let config = write_config(temp.path(), "export", "managed", "sync");
+ let output_path = temp.path().join("out");
+ let output = yy(&[
+ "--volume",
+ config.to_str().unwrap(),
+ "sync",
+ "export",
+ "--output",
+ output_path.to_str().unwrap(),
+ ]);
+ assert!(!output.status.success());
+ assert_stderr(&output, "requires exactly one of --artifact or --conflict");
+}
+
+#[test]
+fn resolve_requires_a_decision() {
+ let temp = tempfile::tempdir().unwrap();
+ let config = write_config(temp.path(), "resolve", "managed", "sync");
+ let output = yy(&[
+ "--volume",
+ config.to_str().unwrap(),
+ "sync",
+ "resolve",
+ "deadbeef",
+ ]);
+ assert!(!output.status.success());
+ assert_stderr(&output, "requires exactly one of --keep or --content");
+}