From bb54f819e5678b2d96b4069df20453008263da01 Mon Sep 17 00:00:00 2001 From: Filipe Forattini Date: Tue, 7 Jul 2026 19:59:09 -0300 Subject: [PATCH 01/19] feat(fork): promote operational fork roots Refs #1780 --- crates/reddb-file/src/operational_manifest.rs | 123 ++++++++++++++++++ 1 file changed, 123 insertions(+) diff --git a/crates/reddb-file/src/operational_manifest.rs b/crates/reddb-file/src/operational_manifest.rs index 2adcdfdaa..b2c584bb9 100644 --- a/crates/reddb-file/src/operational_manifest.rs +++ b/crates/reddb-file/src/operational_manifest.rs @@ -56,6 +56,13 @@ pub struct ForkInfo { pub hydrated: u64, } +#[derive(Debug, Clone)] +pub struct PromoteForkOutcome { + pub name: String, + pub fork_lsn: u64, + pub archived_parent: OperationalManifest, +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ForkHydrationState { SharedByReference, @@ -690,6 +697,70 @@ impl OperationalManifest { Ok(Some(detached)) } + /// Promote a store fork to the primary operational root. + /// + /// The promoted fork is first hydrated through the same materialization path + /// used by restore/fork detach. The superseded primary is then moved to a + /// deterministic retired root, so its disposition is explicit and cannot be + /// mistaken for the active store. + pub fn promote_fork(&self, name: &str) -> io::Result> { + let fork = self.fork_handle(name); + if !fork.root.exists() { + return Ok(None); + } + let origin = fork + .fork_origin()? + .ok_or_else(|| invalid_data(format!("store fork is missing origin: {name}")))?; + if origin.parent_store != self.store_identity() { + return Err(invalid_data(format!( + "store fork {name} belongs to {}, not {}", + origin.parent_store, + self.store_identity() + ))); + } + + let staging = self.promoting_fork_handle(name); + if staging.root.exists() { + return Err(io::Error::new( + io::ErrorKind::AlreadyExists, + format!("store fork promotion staging path already exists: {name}"), + )); + } + let archived_parent = self.archived_parent_handle(name); + if archived_parent.root.exists() { + return Err(io::Error::new( + io::ErrorKind::AlreadyExists, + format!("retired parent store already exists for promoted fork: {name}"), + )); + } + + fork.hydrate_shared_collections()?; + if let Some(parent) = staging.root.parent() { + fs::create_dir_all(parent)?; + } + fs::rename(&fork.root, &staging.root)?; + sync_dir(&self.forks_dir())?; + if let Some(parent) = staging.root.parent() { + sync_dir(parent)?; + } + + fs::rename(&self.root, &archived_parent.root)?; + if let Some(parent) = self.root.parent() { + sync_dir(parent)?; + } + fs::rename(&staging.root, &self.root)?; + if let Some(parent) = self.root.parent() { + sync_dir(parent)?; + } + self.clear_fork_origin()?; + + Ok(Some(PromoteForkOutcome { + name: origin.name, + fork_lsn: origin.fork_lsn, + archived_parent, + })) + } + /// Read this manifest's fork origin, if it is a fork. pub fn fork_origin(&self) -> io::Result> { Ok(self @@ -730,6 +801,34 @@ impl OperationalManifest { } } + fn archived_parent_handle(&self, name: &str) -> Self { + let root_name = self + .root + .file_name() + .map(|name| name.to_string_lossy().into_owned()) + .unwrap_or_else(|| "store.ops".to_string()); + Self { + root: self.root.with_file_name(format!( + "{root_name}.retired-by-promote-{}", + sanitize_component(name) + )), + } + } + + fn promoting_fork_handle(&self, name: &str) -> Self { + let root_name = self + .root + .file_name() + .map(|name| name.to_string_lossy().into_owned()) + .unwrap_or_else(|| "store.ops".to_string()); + Self { + root: self.root.with_file_name(format!( + "{root_name}.promoting-{}", + sanitize_component(name) + )), + } + } + fn hydrate_shared_collections(&self) -> io::Result<()> { let mut manifest = match self.load_current()? { Some(manifest) => manifest, @@ -1891,6 +1990,30 @@ mod tests { ); } + #[test] + fn promote_fork_replaces_primary_and_archives_parent() { + let path = temp_db_path("promote_fork"); + let parent = OperationalManifest::for_db_path(&path); + parent.recover_or_bootstrap(&["users".to_string()]).unwrap(); + write_collection(&parent, "users", b"old-primary"); + parent.create_fork("exp", 21).unwrap(); + let fork = parent.fork_handle("exp"); + fork.hydrate_collection("users").unwrap(); + write_collection(&fork, "users", b"new-primary"); + + let outcome = parent.promote_fork("exp").unwrap().unwrap(); + + assert_eq!(outcome.name, "exp"); + assert_eq!(outcome.fork_lsn, 21); + assert_eq!(read_collection(&parent, "users"), b"new-primary"); + assert!(parent.fork_origin().unwrap().is_none()); + assert!(parent.list_forks().unwrap().is_empty()); + assert_eq!( + read_collection(&outcome.archived_parent, "users"), + b"old-primary" + ); + } + #[test] fn detach_fork_resumes_after_move_before_origin_clear() { let path = temp_db_path("detach_fork_resume"); From 552ab34f98026717903c0e99a1f6ed7472c3f425 Mon Sep 17 00:00:00 2001 From: Filipe Forattini Date: Tue, 7 Jul 2026 20:03:13 -0300 Subject: [PATCH 02/19] feat(rql): add promote fork query Refs #1780 --- crates/reddb-rql/src/core.rs | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/crates/reddb-rql/src/core.rs b/crates/reddb-rql/src/core.rs index 25216c21a..28208ae53 100644 --- a/crates/reddb-rql/src/core.rs +++ b/crates/reddb-rql/src/core.rs @@ -57,6 +57,8 @@ pub enum QueryExpr { DropVcsRef(DropVcsRefQuery), /// FORK STORE AS name [AT LSN n] ForkStore(ForkStoreQuery), + /// PROMOTE FORK name + PromoteFork(PromoteForkQuery), /// DROP FORK name DropFork(DropForkQuery), /// GRAPH subcommand (NEIGHBORHOOD, SHORTEST_PATH, etc.) @@ -2287,6 +2289,12 @@ pub struct ForkStoreQuery { pub at_lsn: Option, } +/// PROMOTE FORK name +#[derive(Debug, Clone)] +pub struct PromoteForkQuery { + pub name: String, +} + /// DROP FORK name #[derive(Debug, Clone)] pub struct DropForkQuery { From e09073be59c8bb83bc0cf55593e27f32e80869e7 Mon Sep 17 00:00:00 2001 From: Filipe Forattini Date: Tue, 7 Jul 2026 20:03:22 -0300 Subject: [PATCH 03/19] feat(rql): parse promote fork Refs #1780 --- crates/reddb-rql/src/sql.rs | 39 ++++++++++++++++++++++++++++++++----- 1 file changed, 34 insertions(+), 5 deletions(-) diff --git a/crates/reddb-rql/src/sql.rs b/crates/reddb-rql/src/sql.rs index 3cb808e57..22c0773d7 100644 --- a/crates/reddb-rql/src/sql.rs +++ b/crates/reddb-rql/src/sql.rs @@ -11,11 +11,11 @@ use crate::ast::{ DropVectorQuery, DropViewQuery, EventsBackfillQuery, ExplainAlterQuery, ExplainMigrationQuery, ExplainQuery, Expr, FieldRef, Filter, ForeignColumnDef, ForkStoreQuery, GrantStmt, GraphCommand, GraphQuery, HybridQuery, InsertQuery, IsolationLevel, JoinQuery, KvCommand, - MaintenanceCommand, PathQuery, PolicyAction, ProbabilisticCommand, QueryExpr, QueueCommand, - QueueSelectQuery, RankOfQuery, RankRangeQuery, RefreshMaterializedViewQuery, RevokeStmt, - RollbackMigrationQuery, SearchCommand, Span, TableQuery, TreeCommand, TruncateQuery, - TxnControl, UpdateQuery, VcsCommand, VcsConflictResolution, VcsRefKind, VcsResetMode, - VectorQuery, + MaintenanceCommand, PathQuery, PolicyAction, ProbabilisticCommand, PromoteForkQuery, QueryExpr, + QueueCommand, QueueSelectQuery, RankOfQuery, RankRangeQuery, RefreshMaterializedViewQuery, + RevokeStmt, RollbackMigrationQuery, SearchCommand, Span, TableQuery, TreeCommand, + TruncateQuery, TxnControl, UpdateQuery, VcsCommand, VcsConflictResolution, VcsRefKind, + VcsResetMode, VectorQuery, }; use crate::lexer::Token; use crate::parser::{ParseError, Parser, SafeTokenDisplay}; @@ -123,6 +123,7 @@ pub enum SqlCommand { Maintenance(MaintenanceCommand), Vcs(VcsCommand), ForkStore(ForkStoreQuery), + PromoteFork(PromoteForkQuery), DropFork(DropForkQuery), CreateSchema(CreateSchemaQuery), DropSchema(DropSchemaQuery), @@ -1301,6 +1302,14 @@ mod tests { )); } + #[test] + fn parse_sql_command_covers_store_fork_promotion() { + let SqlCommand::PromoteFork(query) = sql_command("PROMOTE FORK exp") else { + panic!("PROMOTE FORK should parse as a store fork promotion"); + }; + assert_eq!(query.name, "exp"); + } + #[test] fn parse_sql_command_covers_iam_and_hypertable_dispatch_edges() { assert!(matches!( @@ -1477,6 +1486,7 @@ pub enum SqlAdminCommand { CreateUser(CreateUserStmt), IamPolicy(QueryExpr), ForkStore(ForkStoreQuery), + PromoteFork(PromoteForkQuery), DropFork(DropForkQuery), } @@ -1590,6 +1600,7 @@ impl SqlStatement { SqlStatement::Admin(SqlAdminCommand::Maintenance(cmd)) => SqlCommand::Maintenance(cmd), SqlStatement::Admin(SqlAdminCommand::Vcs(cmd)) => SqlCommand::Vcs(cmd), SqlStatement::Admin(SqlAdminCommand::ForkStore(q)) => SqlCommand::ForkStore(q), + SqlStatement::Admin(SqlAdminCommand::PromoteFork(q)) => SqlCommand::PromoteFork(q), SqlStatement::Admin(SqlAdminCommand::DropFork(q)) => SqlCommand::DropFork(q), SqlStatement::Schema(SqlSchemaCommand::CreateSchema(q)) => SqlCommand::CreateSchema(q), SqlStatement::Schema(SqlSchemaCommand::DropSchema(q)) => SqlCommand::DropSchema(q), @@ -1742,6 +1753,7 @@ impl SqlCommand { SqlCommand::Maintenance(cmd) => QueryExpr::MaintenanceCommand(cmd), SqlCommand::Vcs(cmd) => QueryExpr::VcsCommand(cmd), SqlCommand::ForkStore(q) => QueryExpr::ForkStore(q), + SqlCommand::PromoteFork(q) => QueryExpr::PromoteFork(q), SqlCommand::DropFork(q) => QueryExpr::DropFork(q), SqlCommand::CreateSchema(q) => QueryExpr::CreateSchema(q), SqlCommand::DropSchema(q) => QueryExpr::DropSchema(q), @@ -1878,6 +1890,7 @@ impl SqlCommand { SqlCommand::Maintenance(cmd) => SqlStatement::Admin(SqlAdminCommand::Maintenance(cmd)), SqlCommand::Vcs(cmd) => SqlStatement::Admin(SqlAdminCommand::Vcs(cmd)), SqlCommand::ForkStore(q) => SqlStatement::Admin(SqlAdminCommand::ForkStore(q)), + SqlCommand::PromoteFork(q) => SqlStatement::Admin(SqlAdminCommand::PromoteFork(q)), SqlCommand::DropFork(q) => SqlStatement::Admin(SqlAdminCommand::DropFork(q)), SqlCommand::CreateSchema(q) => SqlStatement::Schema(SqlSchemaCommand::CreateSchema(q)), SqlCommand::DropSchema(q) => SqlStatement::Schema(SqlSchemaCommand::DropSchema(q)), @@ -2848,6 +2861,19 @@ impl<'a> Parser<'a> { Ok(SqlCommand::ForkStore(ForkStoreQuery { name, at_lsn })) } + fn parse_promote_fork_command(&mut self) -> Result { + self.advance()?; // PROMOTE + if !self.consume_ident_ci("FORK")? { + return Err(ParseError::expected( + vec!["FORK"], + self.peek(), + self.position(), + )); + } + let name = self.expect_ident()?; + Ok(SqlCommand::PromoteFork(PromoteForkQuery { name })) + } + fn parse_dotted_admin_path(&mut self, lowercase: bool) -> Result { let mut path = self.expect_ident()?; while self.consume(&Token::Dot)? { @@ -3452,6 +3478,9 @@ impl<'a> Parser<'a> { Token::Ident(name) if name.eq_ignore_ascii_case("FORK") => { self.parse_fork_store_command() } + Token::Ident(name) if name.eq_ignore_ascii_case("PROMOTE") => { + self.parse_promote_fork_command() + } Token::Select => match self.parse_select_query()? { QueryExpr::Table(query) => Ok(SqlCommand::Select(query)), QueryExpr::Join(query) => Ok(SqlCommand::Join(query)), From 9122a85e585dfb4e1d39a94583edd382bf2580cf Mon Sep 17 00:00:00 2001 From: Filipe Forattini Date: Tue, 7 Jul 2026 20:03:31 -0300 Subject: [PATCH 04/19] chore(rql): carry promote fork through builders Refs #1780 --- crates/reddb-rql/src/builders.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/crates/reddb-rql/src/builders.rs b/crates/reddb-rql/src/builders.rs index 4476709ed..06c282722 100644 --- a/crates/reddb-rql/src/builders.rs +++ b/crates/reddb-rql/src/builders.rs @@ -270,6 +270,7 @@ impl JoinQueryBuilder { | QueryExpr::CreateVcsRef(_) | QueryExpr::DropVcsRef(_) | QueryExpr::ForkStore(_) + | QueryExpr::PromoteFork(_) | QueryExpr::DropFork(_) | QueryExpr::GraphCommand(_) | QueryExpr::SearchCommand(_) From 71387326c2fd47bee63ba928fbd4d3f9aa735de7 Mon Sep 17 00:00:00 2001 From: Filipe Forattini Date: Tue, 7 Jul 2026 20:03:39 -0300 Subject: [PATCH 05/19] chore(rql): cost promote fork commands Refs #1780 --- crates/reddb-rql/src/planner/optimizer.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/crates/reddb-rql/src/planner/optimizer.rs b/crates/reddb-rql/src/planner/optimizer.rs index de9b19f87..93403f54b 100644 --- a/crates/reddb-rql/src/planner/optimizer.rs +++ b/crates/reddb-rql/src/planner/optimizer.rs @@ -301,6 +301,7 @@ impl JoinReorderingPass { | QueryExpr::CreateVcsRef(_) | QueryExpr::DropVcsRef(_) | QueryExpr::ForkStore(_) + | QueryExpr::PromoteFork(_) | QueryExpr::DropFork(_) | QueryExpr::GraphCommand(_) | QueryExpr::SearchCommand(_) From 0abb978befcee47dbe888d633cffbadd592328ab Mon Sep 17 00:00:00 2001 From: Filipe Forattini Date: Tue, 7 Jul 2026 20:03:47 -0300 Subject: [PATCH 06/19] chore(rql): pass through promote fork Refs #1780 --- crates/reddb-rql/src/planner/rewriter.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/crates/reddb-rql/src/planner/rewriter.rs b/crates/reddb-rql/src/planner/rewriter.rs index a6fd91312..6daf16754 100644 --- a/crates/reddb-rql/src/planner/rewriter.rs +++ b/crates/reddb-rql/src/planner/rewriter.rs @@ -203,6 +203,7 @@ impl RewriteRule for NormalizeRule { | QueryExpr::CreateVcsRef(_) | QueryExpr::DropVcsRef(_) | QueryExpr::ForkStore(_) + | QueryExpr::PromoteFork(_) | QueryExpr::DropFork(_) | QueryExpr::GraphCommand(_) | QueryExpr::SearchCommand(_) @@ -348,6 +349,7 @@ impl RewriteRule for SimplifyFiltersRule { | QueryExpr::CreateVcsRef(_) | QueryExpr::DropVcsRef(_) | QueryExpr::ForkStore(_) + | QueryExpr::PromoteFork(_) | QueryExpr::DropFork(_) | QueryExpr::GraphCommand(_) | QueryExpr::SearchCommand(_) @@ -449,6 +451,7 @@ impl RewriteRule for SimplifyFiltersRule { | QueryExpr::CreateVcsRef(_) | QueryExpr::DropVcsRef(_) | QueryExpr::ForkStore(_) + | QueryExpr::PromoteFork(_) | QueryExpr::DropFork(_) | QueryExpr::GraphCommand(_) | QueryExpr::SearchCommand(_) From 9b42339847caaf57a8a8daf4daad7131c6b90c35 Mon Sep 17 00:00:00 2001 From: Filipe Forattini Date: Tue, 7 Jul 2026 20:16:22 -0300 Subject: [PATCH 07/19] feat(rql): detect promote fork as sql Refs #1780 --- crates/reddb-rql/src/modes/detect.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/crates/reddb-rql/src/modes/detect.rs b/crates/reddb-rql/src/modes/detect.rs index aac929128..44f7f2429 100644 --- a/crates/reddb-rql/src/modes/detect.rs +++ b/crates/reddb-rql/src/modes/detect.rs @@ -75,6 +75,7 @@ pub fn detect_mode(input: &str) -> QueryMode { | "revert" | "resolve" | "fork" + | "promote" | "copy" | "refresh" | "explain" From a0d10d63d5155b5245caba7d24150c5039e49a9a Mon Sep 17 00:00:00 2001 From: Filipe Forattini Date: Tue, 7 Jul 2026 20:16:31 -0300 Subject: [PATCH 08/19] feat(rql): route promote fork frontend Refs #1780 --- crates/reddb-rql/src/sql.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/crates/reddb-rql/src/sql.rs b/crates/reddb-rql/src/sql.rs index 22c0773d7..c048fb32f 100644 --- a/crates/reddb-rql/src/sql.rs +++ b/crates/reddb-rql/src/sql.rs @@ -2127,7 +2127,9 @@ impl<'a> Parser<'a> { Token::Ident(name) if name.eq_ignore_ascii_case("RESET") => { self.parse_sql_statement().map(FrontendStatement::Sql) } - Token::Ident(name) if name.eq_ignore_ascii_case("FORK") => { + Token::Ident(name) + if name.eq_ignore_ascii_case("FORK") || name.eq_ignore_ascii_case("PROMOTE") => + { self.parse_sql_statement().map(FrontendStatement::Sql) } Token::Ident(name) From ccb2102027ad02013ba61deea59952428902aa9a Mon Sep 17 00:00:00 2001 From: Filipe Forattini Date: Tue, 7 Jul 2026 20:16:40 -0300 Subject: [PATCH 09/19] feat(runtime): import promote fork query Refs #1780 --- crates/reddb-server/src/runtime.rs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/crates/reddb-server/src/runtime.rs b/crates/reddb-server/src/runtime.rs index e40580e7a..dcb2cf641 100644 --- a/crates/reddb-server/src/runtime.rs +++ b/crates/reddb-server/src/runtime.rs @@ -34,8 +34,9 @@ use crate::storage::query::ast::{ DropTimeSeriesQuery, DropTreeQuery, DropVectorQuery, EventsBackfillQuery, ExplainAlterQuery, ExplainFormat, FieldRef, Filter, ForkStoreQuery, FusionStrategy, GraphCommand, HybridQuery, IndexMethod, InsertEntityType, InsertQuery, JoinQuery, JoinType, OrderByClause, - ProbabilisticCommand, Projection, QueryExpr, QueueCommand, QueueSelectQuery, QueueSide, - SearchCommand, TableQuery, TreeCommand, TruncateQuery, UpdateQuery, VectorQuery, VectorSource, + ProbabilisticCommand, Projection, PromoteForkQuery, QueryExpr, QueueCommand, QueueSelectQuery, + QueueSide, SearchCommand, TableQuery, TreeCommand, TruncateQuery, UpdateQuery, VectorQuery, + VectorSource, }; use crate::storage::query::is_universal_entity_source as is_universal_query_source; use crate::storage::query::modes::{detect_mode, parse_multi, QueryMode}; From a7bb790358159b6be4ffd276ee664740755aae1d Mon Sep 17 00:00:00 2001 From: Filipe Forattini Date: Tue, 7 Jul 2026 20:16:49 -0300 Subject: [PATCH 10/19] feat(runtime): dispatch promote fork Refs #1780 --- crates/reddb-server/src/runtime/impl_core.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/crates/reddb-server/src/runtime/impl_core.rs b/crates/reddb-server/src/runtime/impl_core.rs index f5aee9b15..87ba3828f 100644 --- a/crates/reddb-server/src/runtime/impl_core.rs +++ b/crates/reddb-server/src/runtime/impl_core.rs @@ -1032,6 +1032,9 @@ impl RedDBRuntime { QueryExpr::CreateVcsRef(ref create) => self.execute_create_vcs_ref(query, create), QueryExpr::DropVcsRef(ref drop_ref) => self.execute_drop_vcs_ref(query, drop_ref), QueryExpr::ForkStore(ref fork) => self.execute_fork_store(query, fork), + QueryExpr::PromoteFork(ref promote_fork) => { + self.execute_promote_fork(query, promote_fork) + } QueryExpr::DropFork(ref drop_fork) => self.execute_drop_fork(query, drop_fork), QueryExpr::ExplainAlter(ref explain) => self.execute_explain_alter(query, explain), // Graph analytics commands From 82feaeb3c7761097ab0d26a10d1a428a90df6398 Mon Sep 17 00:00:00 2001 From: Filipe Forattini Date: Tue, 7 Jul 2026 20:16:56 -0300 Subject: [PATCH 11/19] feat(runtime): execute promote fork Refs #1780 --- crates/reddb-server/src/runtime/impl_ddl.rs | 28 +++++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/crates/reddb-server/src/runtime/impl_ddl.rs b/crates/reddb-server/src/runtime/impl_ddl.rs index 00fd1a8cc..722be558f 100644 --- a/crates/reddb-server/src/runtime/impl_ddl.rs +++ b/crates/reddb-server/src/runtime/impl_ddl.rs @@ -1436,6 +1436,34 @@ impl RedDBRuntime { )) } + pub fn execute_promote_fork( + &self, + raw_query: &str, + query: &PromoteForkQuery, + ) -> RedDBResult { + self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?; + let path = self.inner.db.path().ok_or_else(|| { + RedDBError::Query("PROMOTE FORK requires a persistent store".to_string()) + })?; + self.flush()?; + let manifest = reddb_file::OperationalManifest::for_db_path(path); + let outcome = manifest + .promote_fork(&query.name) + .map_err(|err| RedDBError::Query(format!("failed to promote store fork: {err}")))?; + let outcome = + outcome.ok_or_else(|| RedDBError::NotFound(format!("store fork '{}'", query.name)))?; + Ok(RuntimeQueryResult::ok_message( + raw_query.to_string(), + &format!( + "store fork '{}' promoted at LSN {}; retired parent archived at {}", + outcome.name, + outcome.fork_lsn, + outcome.archived_parent.store_identity() + ), + "promote_fork", + )) + } + /// Execute EXPLAIN ALTER FOR CREATE TABLE /// /// Pure read: computes the schema diff between the target table's From 97e22206a220008f862078e2fc5f3f0c1692e57d Mon Sep 17 00:00:00 2001 From: Filipe Forattini Date: Tue, 7 Jul 2026 20:17:07 -0300 Subject: [PATCH 12/19] feat(runtime): classify promote as ddl Refs #1780 --- crates/reddb-server/src/runtime/statement_frame.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/reddb-server/src/runtime/statement_frame.rs b/crates/reddb-server/src/runtime/statement_frame.rs index f73e974c2..be55365c8 100644 --- a/crates/reddb-server/src/runtime/statement_frame.rs +++ b/crates/reddb-server/src/runtime/statement_frame.rs @@ -216,7 +216,7 @@ fn statement_kind(query: &str) -> &'static str { | b"APPROX" | b"APPROXIMATE" | b"ZRANK" | b"ZRANGE" | b"LIST" | b"WATCH" | b"GET" | b"HISTORY" => "read", b"INSERT" | b"UPDATE" | b"DELETE" | b"UPSERT" | b"MERGE" | b"COPY" | b"TRUNCATE" => "write", - b"CREATE" | b"ALTER" | b"DROP" | b"REINDEX" | b"VACUUM" | b"ANALYZE" => "ddl", + b"CREATE" | b"ALTER" | b"DROP" | b"PROMOTE" | b"REINDEX" | b"VACUUM" | b"ANALYZE" => "ddl", b"GRANT" | b"REVOKE" => "admin", b"BEGIN" | b"START" | b"COMMIT" | b"ROLLBACK" | b"SAVEPOINT" | b"RELEASE" | b"END" | b"SET" | b"RESET" | b"PREPARE" | b"EXECUTE" | b"DEALLOCATE" | b"USE" => "control", From 55785eb5ac5a7fcbf72bdd3fcd97c48f26ece398 Mon Sep 17 00:00:00 2001 From: Filipe Forattini Date: Tue, 7 Jul 2026 20:17:16 -0300 Subject: [PATCH 13/19] chore(query): name promote fork command Refs #1780 --- crates/reddb-server/src/storage/query/executors/vector.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/crates/reddb-server/src/storage/query/executors/vector.rs b/crates/reddb-server/src/storage/query/executors/vector.rs index 305221171..c6fb6c21d 100644 --- a/crates/reddb-server/src/storage/query/executors/vector.rs +++ b/crates/reddb-server/src/storage/query/executors/vector.rs @@ -586,6 +586,7 @@ fn query_expr_name(expr: &QueryExpr) -> &'static str { QueryExpr::CreateVcsRef(_) => "create_vcs_ref", QueryExpr::DropVcsRef(_) => "drop_vcs_ref", QueryExpr::ForkStore(_) => "fork_store", + QueryExpr::PromoteFork(_) => "promote_fork", QueryExpr::DropFork(_) => "drop_fork", QueryExpr::VcsCommand(_) => "vcs_command", QueryExpr::GraphCommand(_) => "graph_command", From 548a92bf17443aaaa0f8ac151f35cf59ca8b4d65 Mon Sep 17 00:00:00 2001 From: Filipe Forattini Date: Tue, 7 Jul 2026 20:17:25 -0300 Subject: [PATCH 14/19] chore(query): cost promote fork command Refs #1780 --- crates/reddb-server/src/storage/query/planner/cost.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/crates/reddb-server/src/storage/query/planner/cost.rs b/crates/reddb-server/src/storage/query/planner/cost.rs index 930dbe650..d7bbae5c8 100644 --- a/crates/reddb-server/src/storage/query/planner/cost.rs +++ b/crates/reddb-server/src/storage/query/planner/cost.rs @@ -306,6 +306,7 @@ impl CostEstimator { | QueryExpr::CreateVcsRef(_) | QueryExpr::DropVcsRef(_) | QueryExpr::ForkStore(_) + | QueryExpr::PromoteFork(_) | QueryExpr::DropFork(_) | QueryExpr::VcsCommand(_) | QueryExpr::GraphCommand(_) @@ -408,6 +409,7 @@ impl CostEstimator { | QueryExpr::CreateVcsRef(_) | QueryExpr::DropVcsRef(_) | QueryExpr::ForkStore(_) + | QueryExpr::PromoteFork(_) | QueryExpr::DropFork(_) | QueryExpr::VcsCommand(_) | QueryExpr::GraphCommand(_) From 3cbf038c685dffedf0e61b7531b3c6796bdb870a Mon Sep 17 00:00:00 2001 From: Filipe Forattini Date: Tue, 7 Jul 2026 20:17:36 -0300 Subject: [PATCH 15/19] chore(query): plan promote fork command Refs #1780 --- crates/reddb-server/src/storage/query/planner/logical.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/crates/reddb-server/src/storage/query/planner/logical.rs b/crates/reddb-server/src/storage/query/planner/logical.rs index e9ba13b96..91a632bc9 100644 --- a/crates/reddb-server/src/storage/query/planner/logical.rs +++ b/crates/reddb-server/src/storage/query/planner/logical.rs @@ -623,6 +623,7 @@ pub(super) fn logical_plan_node_with_catalog(db: &RedDB, expr: &QueryExpr) -> Ca | QueryExpr::CreateVcsRef(_) | QueryExpr::DropVcsRef(_) | QueryExpr::ForkStore(_) + | QueryExpr::PromoteFork(_) | QueryExpr::DropFork(_) | QueryExpr::VcsCommand(_) | QueryExpr::GraphCommand(_) From fcff34eac735e45cfae519928d3efb0fcaa57dff Mon Sep 17 00:00:00 2001 From: Filipe Forattini Date: Tue, 7 Jul 2026 20:17:48 -0300 Subject: [PATCH 16/19] chore(query): classify promote fork helpers Refs #1780 --- .../reddb-server/src/storage/query/planner/logical_helpers.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/crates/reddb-server/src/storage/query/planner/logical_helpers.rs b/crates/reddb-server/src/storage/query/planner/logical_helpers.rs index 693a86808..42bf58350 100644 --- a/crates/reddb-server/src/storage/query/planner/logical_helpers.rs +++ b/crates/reddb-server/src/storage/query/planner/logical_helpers.rs @@ -444,6 +444,7 @@ pub(crate) fn join_expr_exposes_field_table(expr: &QueryExpr, table: &str) -> bo | QueryExpr::CreateVcsRef(_) | QueryExpr::DropVcsRef(_) | QueryExpr::ForkStore(_) + | QueryExpr::PromoteFork(_) | QueryExpr::DropFork(_) | QueryExpr::VcsCommand(_) | QueryExpr::GraphCommand(_) @@ -829,6 +830,7 @@ pub(crate) fn query_expr_kind(expr: &QueryExpr) -> &'static str { QueryExpr::CreateVcsRef(_) => "create_vcs_ref", QueryExpr::DropVcsRef(_) => "drop_vcs_ref", QueryExpr::ForkStore(_) => "fork_store", + QueryExpr::PromoteFork(_) => "promote_fork", QueryExpr::DropFork(_) => "drop_fork", QueryExpr::VcsCommand(_) => "vcs_command", QueryExpr::GraphCommand(_) => "graph_command", From d45674f55439530d169f30af7806e36320d62012 Mon Sep 17 00:00:00 2001 From: Filipe Forattini Date: Tue, 7 Jul 2026 20:17:58 -0300 Subject: [PATCH 17/19] chore(query): pass promote fork through unified executor Refs #1780 --- crates/reddb-server/src/storage/query/unified/executor.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/crates/reddb-server/src/storage/query/unified/executor.rs b/crates/reddb-server/src/storage/query/unified/executor.rs index 745aa760b..c62f55efd 100644 --- a/crates/reddb-server/src/storage/query/unified/executor.rs +++ b/crates/reddb-server/src/storage/query/unified/executor.rs @@ -190,6 +190,7 @@ impl UnifiedExecutor { | QueryExpr::CreateVcsRef(_) | QueryExpr::DropVcsRef(_) | QueryExpr::ForkStore(_) + | QueryExpr::PromoteFork(_) | QueryExpr::DropFork(_) | QueryExpr::VcsCommand(_) | QueryExpr::GraphCommand(_) @@ -441,6 +442,7 @@ impl UnifiedExecutor { | QueryExpr::CreateVcsRef(_) | QueryExpr::DropVcsRef(_) | QueryExpr::ForkStore(_) + | QueryExpr::PromoteFork(_) | QueryExpr::DropFork(_) | QueryExpr::VcsCommand(_) | QueryExpr::GraphCommand(_) From 8dd5901266a1c1db5daad1a0b2c308dae5aa5457 Mon Sep 17 00:00:00 2001 From: Filipe Forattini Date: Tue, 7 Jul 2026 20:18:08 -0300 Subject: [PATCH 18/19] test(runtime): cover promote fork sql Refs #1780 --- .../reddb-server/tests/store_fork_profile.rs | 42 +++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/crates/reddb-server/tests/store_fork_profile.rs b/crates/reddb-server/tests/store_fork_profile.rs index 0fe5c126c..2d04343fa 100644 --- a/crates/reddb-server/tests/store_fork_profile.rs +++ b/crates/reddb-server/tests/store_fork_profile.rs @@ -1,4 +1,6 @@ +use reddb_file::OperationalManifest; use reddb_server::{RedDBError, RedDBOptions, RedDBRuntime, StorageDeployPreset}; +use reddb_types::Value; fn temp_data_path(name: &str) -> (tempfile::TempDir, std::path::PathBuf) { let dir = tempfile::Builder::new() @@ -86,3 +88,43 @@ fn operational_directory_fork_uses_exported_layout() { .detach_fork_store("experiment") .expect("detach missing fork is idempotent")); } + +#[test] +fn promote_fork_sql_installs_fork_as_primary_and_archives_parent() { + let (_dir, path) = temp_data_path("promote-fork-sql"); + let runtime = RedDBRuntime::with_options(RedDBOptions::persistent(&path)).expect("runtime"); + runtime + .execute_query("CREATE TABLE users (id INT)") + .expect("create table"); + runtime + .execute_query("FORK STORE AS experiment") + .expect("fork store"); + + let manifest = OperationalManifest::for_db_path(&path); + let fork = manifest.fork_handle("experiment"); + fork.hydrate_collection("users").expect("hydrate fork"); + std::fs::write(fork.collection_path_for_test("users"), b"fork-side-write") + .expect("write fork collection"); + + let promoted = runtime + .execute_query("PROMOTE FORK experiment") + .expect("promote fork"); + let message = match promoted.result.records[0].get("message") { + Some(Value::Text(text)) => text.as_ref(), + other => panic!("unexpected promotion message: {other:?}"), + }; + + assert!( + message.contains("retired parent archived at"), + "promotion must report explicit retired-parent disposition: {:?}", + message + ); + assert_eq!( + std::fs::read(manifest.collection_path_for_test("users")).expect("read primary"), + b"fork-side-write" + ); + assert!( + manifest.list_forks().expect("list forks").is_empty(), + "promoted fork must no longer remain a live child fork" + ); +} From 0918d00bc5186189f2f37593562058fe04f795e1 Mon Sep 17 00:00:00 2001 From: Filipe Forattini Date: Tue, 7 Jul 2026 22:25:59 -0300 Subject: [PATCH 19/19] refactor(#1780): extract fork lifecycle into operational_manifest/fork.rs to satisfy 2000-line file-contract guard Claude-Session: https://claude.ai/code/session_0156URehfHzsvrbeAwzGdbCy --- .../src/operational_manifest/fork.rs | 532 ++++++++++++++++++ .../mod.rs} | 516 +---------------- 2 files changed, 538 insertions(+), 510 deletions(-) create mode 100644 crates/reddb-file/src/operational_manifest/fork.rs rename crates/reddb-file/src/{operational_manifest.rs => operational_manifest/mod.rs} (75%) diff --git a/crates/reddb-file/src/operational_manifest/fork.rs b/crates/reddb-file/src/operational_manifest/fork.rs new file mode 100644 index 000000000..b5ecab30d --- /dev/null +++ b/crates/reddb-file/src/operational_manifest/fork.rs @@ -0,0 +1,532 @@ +//! Store-fork lifecycle for the operational manifest (ADR 0070). +//! +//! Fork create/drop/detach/promote, copy-on-write hydration, and the fork +//! listing live here. These are the *store fork* surface — storage mechanics for +//! experiment-and-discard workflows — deliberately distinct from the VCS data +//! model's `branch`/`CHECKPOINT` vocabulary (#1567). +//! +//! The methods are inherent-impl blocks on [`OperationalManifest`], split out +//! from the parent module so each file-contract file stays under the layout +//! authority's per-file line budget. Behavior is identical to the pre-split code. + +use std::collections::{BTreeMap, BTreeSet}; +use std::fs; +use std::io; +use std::path::{Path, PathBuf}; + +use super::{ + copy_file_durable, invalid_data, sanitize_component, sync_dir, CollectionEntry, + CollectionState, Manifest, OperationalManifest, FORKS_DIR, +}; + +/// Where a store fork came from: the parent store's identity and the durable LSN +/// the fork is pinned at. Recorded in the fork's own operational manifest so a +/// listing can report each fork's parent and fork LSN without opening the parent. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ForkOrigin { + /// The fork's own name (as given to `FORK STORE AS `). + pub name: String, + /// Identity of the parent store this fork was taken from. + pub parent_store: String, + /// The durable LSN the fork is pinned at (the parent's current durable LSN + /// at fork-create time). + pub fork_lsn: u64, +} + +/// A single row of the fork listing (`SHOW FORKS`): the fork name plus origin. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ForkInfo { + pub name: String, + pub parent_store: String, + pub fork_lsn: u64, + pub hydration_state: ForkHydrationState, + pub collections_total: u64, + pub shared_by_reference: u64, + pub hydrating: u64, + pub hydrated: u64, +} + +#[derive(Debug, Clone)] +pub struct PromoteForkOutcome { + pub name: String, + pub fork_lsn: u64, + pub archived_parent: OperationalManifest, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ForkHydrationState { + SharedByReference, + Hydrating, + Hydrated, +} + +impl ForkHydrationState { + pub fn as_str(self) -> &'static str { + match self { + Self::SharedByReference => "shared_by_reference", + Self::Hydrating => "hydrating", + Self::Hydrated => "hydrated", + } + } +} + +#[derive(Debug, Clone, Copy)] +struct ForkHydrationProgress { + state: ForkHydrationState, + collections_total: u64, + shared_by_reference: u64, + hydrating: u64, + hydrated: u64, +} + +impl Default for ForkHydrationProgress { + fn default() -> Self { + Self { + state: ForkHydrationState::Hydrated, + collections_total: 0, + shared_by_reference: 0, + hydrating: 0, + hydrated: 0, + } + } +} + +impl OperationalManifest { + /// A fork's own operational manifest, rooted under this store's `forks/` dir. + /// The fork is a full operational store root in its own right. + pub fn fork_handle(&self, name: &str) -> Self { + Self { + root: self.forks_dir().join(sanitize_component(name)), + } + } + + /// Create a store fork pinned at `fork_lsn` (ADR 0070). O(metadata): every + /// active parent collection is referenced by absolute source path — no data + /// file is copied at create time. The fork gets its own operational manifest + /// carrying [`ForkOrigin`]; mutable collection files hydrate lazily on first + /// write (see [`hydrate_collection`](Self::hydrate_collection)). + /// + /// This is the *store fork* surface, distinct from the VCS `CHECKPOINT`/branch + /// model (#1567): forks live on the storage/deploy axis. + pub fn create_fork(&self, name: &str, fork_lsn: u64) -> io::Result<()> { + let parent = self + .load_current()? + .ok_or_else(|| invalid_data("cannot fork a store with no operational manifest"))?; + let fork = self.fork_handle(name); + if fork.load_current()?.is_some() { + return Err(io::Error::new( + io::ErrorKind::AlreadyExists, + format!("store fork already exists: {name}"), + )); + } + fork.ensure_dirs()?; + + let mut collections = BTreeMap::new(); + for (cname, entry) in &parent.collections { + if entry.state != CollectionState::Active { + continue; + } + let source = self.collections_dir().join(&entry.path); + collections.insert( + cname.clone(), + CollectionEntry { + state: CollectionState::Active, + path: entry.path.clone(), + source: Some(source.to_string_lossy().into_owned()), + }, + ); + } + + let manifest = Manifest { + generation: 0, + collections, + append_only_segments: Vec::new(), + append_only_retired: BTreeMap::new(), + fork_origin: Some(ForkOrigin { + name: name.to_string(), + parent_store: self.store_identity(), + fork_lsn, + }), + }; + fork.publish(&manifest)?; + sync_dir(&self.forks_dir()) + } + + /// Hydrate a fork collection's private copy (copy-on-write): called on the + /// *fork* handle before the fork first writes a shared collection. Copies the + /// referenced parent bytes into the fork's own `collections/` dir and drops + /// the shared reference. Idempotent and a no-op for owned collections. + pub fn hydrate_collection(&self, name: &str) -> io::Result<()> { + let mut manifest = match self.load_current()? { + Some(manifest) => manifest, + None => return Ok(()), + }; + let Some(entry) = manifest.collections.get_mut(name) else { + return Ok(()); + }; + let Some(source) = entry.source.clone() else { + return Ok(()); + }; + self.ensure_dirs()?; + let dest = self.collections_dir().join(&entry.path); + copy_file_durable(Path::new(&source), &dest)?; + entry.source = None; + manifest.generation += 1; + self.publish(&manifest) + } + + /// Preserve fork isolation before the *parent* mutates a collection file in + /// place: any live fork still sharing that collection by reference gets its + /// as-of-fork snapshot materialized first (copy-on-write on the parent side). + /// Call on the parent handle immediately before an in-place collection write. + pub fn materialize_forks_before_write(&self, collection: &str) -> io::Result<()> { + let parent = match self.load_current()? { + Some(manifest) => manifest, + None => return Ok(()), + }; + let Some(entry) = parent.collections.get(collection) else { + return Ok(()); + }; + let source = self.collections_dir().join(&entry.path); + for fork in self.fork_handles()? { + let mut manifest = match fork.load_current()? { + Some(manifest) => manifest, + None => continue, + }; + let Some(fork_entry) = manifest.collections.get_mut(collection) else { + continue; + }; + if fork_entry.source.is_none() { + continue; + } + fork.ensure_dirs()?; + let dest = fork.collections_dir().join(&fork_entry.path); + copy_file_durable(&source, &dest)?; + fork_entry.source = None; + manifest.generation += 1; + fork.publish(&manifest)?; + } + Ok(()) + } + + /// List every store fork of this store: name, parent identity, and fork LSN. + pub fn list_forks(&self) -> io::Result> { + let mut out = Vec::new(); + let entries = match fs::read_dir(self.forks_dir()) { + Ok(entries) => entries, + Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(out), + Err(err) => return Err(err), + }; + for entry in entries { + let entry = entry?; + if !entry.file_type()?.is_dir() { + continue; + } + let fork = Self { root: entry.path() }; + if let Some(manifest) = fork.load_current()? { + let Some(origin) = manifest.fork_origin.as_ref() else { + continue; + }; + let progress = fork.hydration_progress(&manifest); + out.push(ForkInfo { + name: origin.name.clone(), + parent_store: origin.parent_store.clone(), + fork_lsn: origin.fork_lsn, + hydration_state: progress.state, + collections_total: progress.collections_total, + shared_by_reference: progress.shared_by_reference, + hydrating: progress.hydrating, + hydrated: progress.hydrated, + }); + } + } + out.sort_by(|a, b| a.name.cmp(&b.name)); + Ok(out) + } + + /// Drop a store fork: remove its manifest and garbage-collect its private + /// (hydrated) files. Shared-by-reference entries are only path strings, so + /// the parent store's data is never touched. Idempotent. Returns whether a + /// fork was actually removed. + pub fn drop_fork(&self, name: &str) -> io::Result { + let fork = self.fork_handle(name); + if !fork.root.exists() { + return Ok(false); + } + fs::remove_dir_all(&fork.root)?; + let _ = sync_dir(&self.forks_dir()); + if let Some(manifest) = self.load_current()? { + let protected_paths = self.protected_collection_paths(&manifest)?; + self.quarantine_unreferenced_collection_files(&protected_paths)?; + self.quarantine_unreferenced_append_only_segments(&manifest)?; + } + Ok(true) + } + + /// Detach a store fork into an independent operational store root. All + /// shared-by-reference collections are hydrated before the fork is moved out + /// from under the parent store; the final manifest then drops its + /// [`ForkOrigin`], so parent retention and WAL pruning no longer see it as a + /// live fork. + /// + /// The operation is restartable across the two durable phases: + /// - if hydration finished but the fork is still nested, rerun hydrates as a + /// no-op and moves it; + /// - if the move finished but origin clearing did not, rerun finishes the + /// detached manifest in place. + pub fn detach_fork(&self, name: &str) -> io::Result> { + let fork = self.fork_handle(name); + let detached = self.detached_fork_handle(name); + + if detached.root.exists() { + if fork.root.exists() { + return Err(io::Error::new( + io::ErrorKind::AlreadyExists, + format!("detached store already exists for fork: {name}"), + )); + } + let was_fork = detached.fork_origin()?.is_some(); + detached.clear_fork_origin()?; + return Ok(was_fork.then_some(detached)); + } + + if !fork.root.exists() { + return Ok(None); + } + + fork.hydrate_shared_collections()?; + if let Some(parent) = detached.root.parent() { + fs::create_dir_all(parent)?; + } + fs::rename(&fork.root, &detached.root)?; + sync_dir(&self.forks_dir())?; + if let Some(parent) = detached.root.parent() { + sync_dir(parent)?; + } + detached.clear_fork_origin()?; + Ok(Some(detached)) + } + + /// Promote a store fork to the primary operational root. + /// + /// The promoted fork is first hydrated through the same materialization path + /// used by restore/fork detach. The superseded primary is then moved to a + /// deterministic retired root, so its disposition is explicit and cannot be + /// mistaken for the active store. + pub fn promote_fork(&self, name: &str) -> io::Result> { + let fork = self.fork_handle(name); + if !fork.root.exists() { + return Ok(None); + } + let origin = fork + .fork_origin()? + .ok_or_else(|| invalid_data(format!("store fork is missing origin: {name}")))?; + if origin.parent_store != self.store_identity() { + return Err(invalid_data(format!( + "store fork {name} belongs to {}, not {}", + origin.parent_store, + self.store_identity() + ))); + } + + let staging = self.promoting_fork_handle(name); + if staging.root.exists() { + return Err(io::Error::new( + io::ErrorKind::AlreadyExists, + format!("store fork promotion staging path already exists: {name}"), + )); + } + let archived_parent = self.archived_parent_handle(name); + if archived_parent.root.exists() { + return Err(io::Error::new( + io::ErrorKind::AlreadyExists, + format!("retired parent store already exists for promoted fork: {name}"), + )); + } + + fork.hydrate_shared_collections()?; + if let Some(parent) = staging.root.parent() { + fs::create_dir_all(parent)?; + } + fs::rename(&fork.root, &staging.root)?; + sync_dir(&self.forks_dir())?; + if let Some(parent) = staging.root.parent() { + sync_dir(parent)?; + } + + fs::rename(&self.root, &archived_parent.root)?; + if let Some(parent) = self.root.parent() { + sync_dir(parent)?; + } + fs::rename(&staging.root, &self.root)?; + if let Some(parent) = self.root.parent() { + sync_dir(parent)?; + } + self.clear_fork_origin()?; + + Ok(Some(PromoteForkOutcome { + name: origin.name, + fork_lsn: origin.fork_lsn, + archived_parent, + })) + } + + /// Read this manifest's fork origin, if it is a fork. + pub fn fork_origin(&self) -> io::Result> { + Ok(self + .load_current()? + .and_then(|manifest| manifest.fork_origin)) + } + + fn forks_dir(&self) -> PathBuf { + self.root.join(FORKS_DIR) + } + + fn fork_handles(&self) -> io::Result> { + let mut out = Vec::new(); + let entries = match fs::read_dir(self.forks_dir()) { + Ok(entries) => entries, + Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(out), + Err(err) => return Err(err), + }; + for entry in entries { + let entry = entry?; + if entry.file_type()?.is_dir() { + out.push(Self { root: entry.path() }); + } + } + Ok(out) + } + + pub(super) fn detached_fork_handle(&self, name: &str) -> Self { + let root_name = self + .root + .file_name() + .map(|name| name.to_string_lossy().into_owned()) + .unwrap_or_else(|| "store.ops".to_string()); + Self { + root: self + .root + .with_file_name(format!("{root_name}.detached-{}", sanitize_component(name))), + } + } + + fn archived_parent_handle(&self, name: &str) -> Self { + let root_name = self + .root + .file_name() + .map(|name| name.to_string_lossy().into_owned()) + .unwrap_or_else(|| "store.ops".to_string()); + Self { + root: self.root.with_file_name(format!( + "{root_name}.retired-by-promote-{}", + sanitize_component(name) + )), + } + } + + fn promoting_fork_handle(&self, name: &str) -> Self { + let root_name = self + .root + .file_name() + .map(|name| name.to_string_lossy().into_owned()) + .unwrap_or_else(|| "store.ops".to_string()); + Self { + root: self.root.with_file_name(format!( + "{root_name}.promoting-{}", + sanitize_component(name) + )), + } + } + + pub(super) fn hydrate_shared_collections(&self) -> io::Result<()> { + let mut manifest = match self.load_current()? { + Some(manifest) => manifest, + None => return Ok(()), + }; + let mut changed = false; + self.ensure_dirs()?; + for entry in manifest.collections.values_mut() { + let Some(source) = entry.source.clone() else { + continue; + }; + let dest = self.collections_dir().join(&entry.path); + copy_file_durable(Path::new(&source), &dest)?; + entry.source = None; + changed = true; + } + if changed { + manifest.generation += 1; + self.publish(&manifest)?; + } + Ok(()) + } + + fn clear_fork_origin(&self) -> io::Result<()> { + let mut manifest = match self.load_current()? { + Some(manifest) => manifest, + None => return Ok(()), + }; + if manifest.fork_origin.is_none() { + return Ok(()); + } + manifest.fork_origin = None; + manifest.generation += 1; + self.publish(&manifest) + } + + pub(super) fn fork_referenced_collection_paths(&self) -> io::Result> { + let mut paths = BTreeSet::new(); + let parent_identity = self.store_identity(); + let parent_collections_dir = self.collections_dir(); + for fork in self.fork_handles()? { + let Some(manifest) = fork.load_current()? else { + continue; + }; + if manifest + .fork_origin + .as_ref() + .map(|origin| origin.parent_store.as_str()) + != Some(parent_identity.as_str()) + { + continue; + } + for entry in manifest.collections.values() { + let Some(source) = &entry.source else { + continue; + }; + if Path::new(source) == parent_collections_dir.join(&entry.path) { + paths.insert(entry.path.clone()); + } + } + } + Ok(paths) + } + + fn hydration_progress(&self, manifest: &Manifest) -> ForkHydrationProgress { + let mut progress = ForkHydrationProgress::default(); + for entry in manifest.collections.values() { + if entry.state != CollectionState::Active { + continue; + } + progress.collections_total += 1; + if entry.source.is_none() { + progress.hydrated += 1; + continue; + } + if self.collections_dir().join(&entry.path).exists() { + progress.hydrating += 1; + } else { + progress.shared_by_reference += 1; + } + } + progress.state = if progress.hydrating > 0 { + ForkHydrationState::Hydrating + } else if progress.shared_by_reference > 0 { + ForkHydrationState::SharedByReference + } else { + ForkHydrationState::Hydrated + }; + progress + } +} diff --git a/crates/reddb-file/src/operational_manifest.rs b/crates/reddb-file/src/operational_manifest/mod.rs similarity index 75% rename from crates/reddb-file/src/operational_manifest.rs rename to crates/reddb-file/src/operational_manifest/mod.rs index b2c584bb9..c6c7e5dc3 100644 --- a/crates/reddb-file/src/operational_manifest.rs +++ b/crates/reddb-file/src/operational_manifest/mod.rs @@ -16,6 +16,12 @@ use crate::append_only_segment::{ AppendOnlySegmentChunkChecksum, AppendOnlySegmentCodec, APPEND_ONLY_SEGMENT_CHUNK_BYTES, }; +mod fork; + +// Fork lifecycle types live in the `fork` submodule; re-export so the public +// path (`operational_manifest::ForkInfo`, etc.) is unchanged. +pub use fork::{ForkHydrationState, ForkInfo, ForkOrigin, PromoteForkOutcome}; + const FORMAT_VERSION: u32 = 1; pub const MANIFEST_FILE: &str = "manifest.json"; pub const NEXT_MANIFEST_FILE: &str = "manifest.json.next"; @@ -29,57 +35,6 @@ pub const QUARANTINE_DIR: &str = "quarantine"; /// storage mechanics for experiment-and-discard workflows; it is never a branch. pub const FORKS_DIR: &str = "forks"; -/// Where a store fork came from: the parent store's identity and the durable LSN -/// the fork is pinned at. Recorded in the fork's own operational manifest so a -/// listing can report each fork's parent and fork LSN without opening the parent. -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct ForkOrigin { - /// The fork's own name (as given to `FORK STORE AS `). - pub name: String, - /// Identity of the parent store this fork was taken from. - pub parent_store: String, - /// The durable LSN the fork is pinned at (the parent's current durable LSN - /// at fork-create time). - pub fork_lsn: u64, -} - -/// A single row of the fork listing (`SHOW FORKS`): the fork name plus origin. -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct ForkInfo { - pub name: String, - pub parent_store: String, - pub fork_lsn: u64, - pub hydration_state: ForkHydrationState, - pub collections_total: u64, - pub shared_by_reference: u64, - pub hydrating: u64, - pub hydrated: u64, -} - -#[derive(Debug, Clone)] -pub struct PromoteForkOutcome { - pub name: String, - pub fork_lsn: u64, - pub archived_parent: OperationalManifest, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum ForkHydrationState { - SharedByReference, - Hydrating, - Hydrated, -} - -impl ForkHydrationState { - pub fn as_str(self) -> &'static str { - match self { - Self::SharedByReference => "shared_by_reference", - Self::Hydrating => "hydrating", - Self::Hydrated => "hydrated", - } - } -} - #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum CollectionState { Active, @@ -334,177 +289,6 @@ impl OperationalManifest { self.root.to_string_lossy().into_owned() } - /// A fork's own operational manifest, rooted under this store's `forks/` dir. - /// The fork is a full operational store root in its own right. - pub fn fork_handle(&self, name: &str) -> Self { - Self { - root: self.forks_dir().join(sanitize_component(name)), - } - } - - /// Create a store fork pinned at `fork_lsn` (ADR 0070). O(metadata): every - /// active parent collection is referenced by absolute source path — no data - /// file is copied at create time. The fork gets its own operational manifest - /// carrying [`ForkOrigin`]; mutable collection files hydrate lazily on first - /// write (see [`hydrate_collection`](Self::hydrate_collection)). - /// - /// This is the *store fork* surface, distinct from the VCS `CHECKPOINT`/branch - /// model (#1567): forks live on the storage/deploy axis. - pub fn create_fork(&self, name: &str, fork_lsn: u64) -> io::Result<()> { - let parent = self - .load_current()? - .ok_or_else(|| invalid_data("cannot fork a store with no operational manifest"))?; - let fork = self.fork_handle(name); - if fork.load_current()?.is_some() { - return Err(io::Error::new( - io::ErrorKind::AlreadyExists, - format!("store fork already exists: {name}"), - )); - } - fork.ensure_dirs()?; - - let mut collections = BTreeMap::new(); - for (cname, entry) in &parent.collections { - if entry.state != CollectionState::Active { - continue; - } - let source = self.collections_dir().join(&entry.path); - collections.insert( - cname.clone(), - CollectionEntry { - state: CollectionState::Active, - path: entry.path.clone(), - source: Some(source.to_string_lossy().into_owned()), - }, - ); - } - - let manifest = Manifest { - generation: 0, - collections, - append_only_segments: Vec::new(), - append_only_retired: BTreeMap::new(), - fork_origin: Some(ForkOrigin { - name: name.to_string(), - parent_store: self.store_identity(), - fork_lsn, - }), - }; - fork.publish(&manifest)?; - sync_dir(&self.forks_dir()) - } - - /// Hydrate a fork collection's private copy (copy-on-write): called on the - /// *fork* handle before the fork first writes a shared collection. Copies the - /// referenced parent bytes into the fork's own `collections/` dir and drops - /// the shared reference. Idempotent and a no-op for owned collections. - pub fn hydrate_collection(&self, name: &str) -> io::Result<()> { - let mut manifest = match self.load_current()? { - Some(manifest) => manifest, - None => return Ok(()), - }; - let Some(entry) = manifest.collections.get_mut(name) else { - return Ok(()); - }; - let Some(source) = entry.source.clone() else { - return Ok(()); - }; - self.ensure_dirs()?; - let dest = self.collections_dir().join(&entry.path); - copy_file_durable(Path::new(&source), &dest)?; - entry.source = None; - manifest.generation += 1; - self.publish(&manifest) - } - - /// Preserve fork isolation before the *parent* mutates a collection file in - /// place: any live fork still sharing that collection by reference gets its - /// as-of-fork snapshot materialized first (copy-on-write on the parent side). - /// Call on the parent handle immediately before an in-place collection write. - pub fn materialize_forks_before_write(&self, collection: &str) -> io::Result<()> { - let parent = match self.load_current()? { - Some(manifest) => manifest, - None => return Ok(()), - }; - let Some(entry) = parent.collections.get(collection) else { - return Ok(()); - }; - let source = self.collections_dir().join(&entry.path); - for fork in self.fork_handles()? { - let mut manifest = match fork.load_current()? { - Some(manifest) => manifest, - None => continue, - }; - let Some(fork_entry) = manifest.collections.get_mut(collection) else { - continue; - }; - if fork_entry.source.is_none() { - continue; - } - fork.ensure_dirs()?; - let dest = fork.collections_dir().join(&fork_entry.path); - copy_file_durable(&source, &dest)?; - fork_entry.source = None; - manifest.generation += 1; - fork.publish(&manifest)?; - } - Ok(()) - } - - /// List every store fork of this store: name, parent identity, and fork LSN. - pub fn list_forks(&self) -> io::Result> { - let mut out = Vec::new(); - let entries = match fs::read_dir(self.forks_dir()) { - Ok(entries) => entries, - Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(out), - Err(err) => return Err(err), - }; - for entry in entries { - let entry = entry?; - if !entry.file_type()?.is_dir() { - continue; - } - let fork = Self { root: entry.path() }; - if let Some(manifest) = fork.load_current()? { - let Some(origin) = manifest.fork_origin.as_ref() else { - continue; - }; - let progress = fork.hydration_progress(&manifest); - out.push(ForkInfo { - name: origin.name.clone(), - parent_store: origin.parent_store.clone(), - fork_lsn: origin.fork_lsn, - hydration_state: progress.state, - collections_total: progress.collections_total, - shared_by_reference: progress.shared_by_reference, - hydrating: progress.hydrating, - hydrated: progress.hydrated, - }); - } - } - out.sort_by(|a, b| a.name.cmp(&b.name)); - Ok(out) - } - - /// Drop a store fork: remove its manifest and garbage-collect its private - /// (hydrated) files. Shared-by-reference entries are only path strings, so - /// the parent store's data is never touched. Idempotent. Returns whether a - /// fork was actually removed. - pub fn drop_fork(&self, name: &str) -> io::Result { - let fork = self.fork_handle(name); - if !fork.root.exists() { - return Ok(false); - } - fs::remove_dir_all(&fork.root)?; - let _ = sync_dir(&self.forks_dir()); - if let Some(manifest) = self.load_current()? { - let protected_paths = self.protected_collection_paths(&manifest)?; - self.quarantine_unreferenced_collection_files(&protected_paths)?; - self.quarantine_unreferenced_append_only_segments(&manifest)?; - } - Ok(true) - } - pub fn publish_append_only_segment( &self, collection: &str, @@ -653,279 +437,12 @@ impl OperationalManifest { Ok(true) } - /// Detach a store fork into an independent operational store root. All - /// shared-by-reference collections are hydrated before the fork is moved out - /// from under the parent store; the final manifest then drops its - /// [`ForkOrigin`], so parent retention and WAL pruning no longer see it as a - /// live fork. - /// - /// The operation is restartable across the two durable phases: - /// - if hydration finished but the fork is still nested, rerun hydrates as a - /// no-op and moves it; - /// - if the move finished but origin clearing did not, rerun finishes the - /// detached manifest in place. - pub fn detach_fork(&self, name: &str) -> io::Result> { - let fork = self.fork_handle(name); - let detached = self.detached_fork_handle(name); - - if detached.root.exists() { - if fork.root.exists() { - return Err(io::Error::new( - io::ErrorKind::AlreadyExists, - format!("detached store already exists for fork: {name}"), - )); - } - let was_fork = detached.fork_origin()?.is_some(); - detached.clear_fork_origin()?; - return Ok(was_fork.then_some(detached)); - } - - if !fork.root.exists() { - return Ok(None); - } - - fork.hydrate_shared_collections()?; - if let Some(parent) = detached.root.parent() { - fs::create_dir_all(parent)?; - } - fs::rename(&fork.root, &detached.root)?; - sync_dir(&self.forks_dir())?; - if let Some(parent) = detached.root.parent() { - sync_dir(parent)?; - } - detached.clear_fork_origin()?; - Ok(Some(detached)) - } - - /// Promote a store fork to the primary operational root. - /// - /// The promoted fork is first hydrated through the same materialization path - /// used by restore/fork detach. The superseded primary is then moved to a - /// deterministic retired root, so its disposition is explicit and cannot be - /// mistaken for the active store. - pub fn promote_fork(&self, name: &str) -> io::Result> { - let fork = self.fork_handle(name); - if !fork.root.exists() { - return Ok(None); - } - let origin = fork - .fork_origin()? - .ok_or_else(|| invalid_data(format!("store fork is missing origin: {name}")))?; - if origin.parent_store != self.store_identity() { - return Err(invalid_data(format!( - "store fork {name} belongs to {}, not {}", - origin.parent_store, - self.store_identity() - ))); - } - - let staging = self.promoting_fork_handle(name); - if staging.root.exists() { - return Err(io::Error::new( - io::ErrorKind::AlreadyExists, - format!("store fork promotion staging path already exists: {name}"), - )); - } - let archived_parent = self.archived_parent_handle(name); - if archived_parent.root.exists() { - return Err(io::Error::new( - io::ErrorKind::AlreadyExists, - format!("retired parent store already exists for promoted fork: {name}"), - )); - } - - fork.hydrate_shared_collections()?; - if let Some(parent) = staging.root.parent() { - fs::create_dir_all(parent)?; - } - fs::rename(&fork.root, &staging.root)?; - sync_dir(&self.forks_dir())?; - if let Some(parent) = staging.root.parent() { - sync_dir(parent)?; - } - - fs::rename(&self.root, &archived_parent.root)?; - if let Some(parent) = self.root.parent() { - sync_dir(parent)?; - } - fs::rename(&staging.root, &self.root)?; - if let Some(parent) = self.root.parent() { - sync_dir(parent)?; - } - self.clear_fork_origin()?; - - Ok(Some(PromoteForkOutcome { - name: origin.name, - fork_lsn: origin.fork_lsn, - archived_parent, - })) - } - - /// Read this manifest's fork origin, if it is a fork. - pub fn fork_origin(&self) -> io::Result> { - Ok(self - .load_current()? - .and_then(|manifest| manifest.fork_origin)) - } - - fn forks_dir(&self) -> PathBuf { - self.root.join(FORKS_DIR) - } - - fn fork_handles(&self) -> io::Result> { - let mut out = Vec::new(); - let entries = match fs::read_dir(self.forks_dir()) { - Ok(entries) => entries, - Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(out), - Err(err) => return Err(err), - }; - for entry in entries { - let entry = entry?; - if entry.file_type()?.is_dir() { - out.push(Self { root: entry.path() }); - } - } - Ok(out) - } - - fn detached_fork_handle(&self, name: &str) -> Self { - let root_name = self - .root - .file_name() - .map(|name| name.to_string_lossy().into_owned()) - .unwrap_or_else(|| "store.ops".to_string()); - Self { - root: self - .root - .with_file_name(format!("{root_name}.detached-{}", sanitize_component(name))), - } - } - - fn archived_parent_handle(&self, name: &str) -> Self { - let root_name = self - .root - .file_name() - .map(|name| name.to_string_lossy().into_owned()) - .unwrap_or_else(|| "store.ops".to_string()); - Self { - root: self.root.with_file_name(format!( - "{root_name}.retired-by-promote-{}", - sanitize_component(name) - )), - } - } - - fn promoting_fork_handle(&self, name: &str) -> Self { - let root_name = self - .root - .file_name() - .map(|name| name.to_string_lossy().into_owned()) - .unwrap_or_else(|| "store.ops".to_string()); - Self { - root: self.root.with_file_name(format!( - "{root_name}.promoting-{}", - sanitize_component(name) - )), - } - } - - fn hydrate_shared_collections(&self) -> io::Result<()> { - let mut manifest = match self.load_current()? { - Some(manifest) => manifest, - None => return Ok(()), - }; - let mut changed = false; - self.ensure_dirs()?; - for entry in manifest.collections.values_mut() { - let Some(source) = entry.source.clone() else { - continue; - }; - let dest = self.collections_dir().join(&entry.path); - copy_file_durable(Path::new(&source), &dest)?; - entry.source = None; - changed = true; - } - if changed { - manifest.generation += 1; - self.publish(&manifest)?; - } - Ok(()) - } - - fn clear_fork_origin(&self) -> io::Result<()> { - let mut manifest = match self.load_current()? { - Some(manifest) => manifest, - None => return Ok(()), - }; - if manifest.fork_origin.is_none() { - return Ok(()); - } - manifest.fork_origin = None; - manifest.generation += 1; - self.publish(&manifest) - } - fn protected_collection_paths(&self, manifest: &Manifest) -> io::Result> { let mut paths = active_collection_paths(manifest); paths.extend(self.fork_referenced_collection_paths()?); Ok(paths) } - fn fork_referenced_collection_paths(&self) -> io::Result> { - let mut paths = BTreeSet::new(); - let parent_identity = self.store_identity(); - let parent_collections_dir = self.collections_dir(); - for fork in self.fork_handles()? { - let Some(manifest) = fork.load_current()? else { - continue; - }; - if manifest - .fork_origin - .as_ref() - .map(|origin| origin.parent_store.as_str()) - != Some(parent_identity.as_str()) - { - continue; - } - for entry in manifest.collections.values() { - let Some(source) = &entry.source else { - continue; - }; - if Path::new(source) == parent_collections_dir.join(&entry.path) { - paths.insert(entry.path.clone()); - } - } - } - Ok(paths) - } - - fn hydration_progress(&self, manifest: &Manifest) -> ForkHydrationProgress { - let mut progress = ForkHydrationProgress::default(); - for entry in manifest.collections.values() { - if entry.state != CollectionState::Active { - continue; - } - progress.collections_total += 1; - if entry.source.is_none() { - progress.hydrated += 1; - continue; - } - if self.collections_dir().join(&entry.path).exists() { - progress.hydrating += 1; - } else { - progress.shared_by_reference += 1; - } - } - progress.state = if progress.hydrating > 0 { - ForkHydrationState::Hydrating - } else if progress.shared_by_reference > 0 { - ForkHydrationState::SharedByReference - } else { - ForkHydrationState::Hydrated - }; - progress - } - pub fn read_generation_for_test(&self) -> io::Result { self.load_current()? .map(|manifest| manifest.generation) @@ -1117,27 +634,6 @@ impl OperationalManifest { } } -#[derive(Debug, Clone, Copy)] -struct ForkHydrationProgress { - state: ForkHydrationState, - collections_total: u64, - shared_by_reference: u64, - hydrating: u64, - hydrated: u64, -} - -impl Default for ForkHydrationProgress { - fn default() -> Self { - Self { - state: ForkHydrationState::Hydrated, - collections_total: 0, - shared_by_reference: 0, - hydrating: 0, - hydrated: 0, - } - } -} - fn empty_manifest() -> Manifest { Manifest { generation: 0,