diff --git a/Cargo.lock b/Cargo.lock index a36795ee0..ba13f620d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2517,52 +2517,6 @@ version = "2.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d7a1e2f27636f116493b8b860f5546edb47c8d8f8ea73e1d2a20be88e28d1fea" -[[package]] -name = "datafusion" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "93db0e623840612f7f2cd757f7e8a8922064192363732c88692e0870016e141b" -dependencies = [ - "arrow", - "arrow-schema", - "async-trait", - "bytes", - "chrono", - "datafusion-catalog 53.1.0", - "datafusion-catalog-listing 53.1.0", - "datafusion-common 53.1.0", - "datafusion-common-runtime 53.1.0", - "datafusion-datasource 53.1.0", - "datafusion-datasource-arrow 53.1.0", - "datafusion-datasource-csv 53.1.0", - "datafusion-datasource-json 53.1.0", - "datafusion-execution 53.1.0", - "datafusion-expr 53.1.0", - "datafusion-expr-common 53.1.0", - "datafusion-functions 53.1.0", - "datafusion-functions-aggregate 53.1.0", - "datafusion-functions-table 53.1.0", - "datafusion-functions-window 53.1.0", - "datafusion-optimizer 53.1.0", - "datafusion-physical-expr 53.1.0", - "datafusion-physical-expr-adapter 53.1.0", - "datafusion-physical-expr-common 53.1.0", - "datafusion-physical-optimizer 53.1.0", - "datafusion-physical-plan 53.1.0", - "datafusion-session 53.1.0", - "futures", - "itertools 0.14.0", - "log", - "object_store", - "parking_lot", - "rand 0.9.4", - "regex", - "tempfile", - "tokio", - "url", - "uuid", -] - [[package]] name = "datafusion" version = "54.1.0" @@ -2573,30 +2527,30 @@ dependencies = [ "arrow-schema", "async-trait", "chrono", - "datafusion-catalog 54.1.0", - "datafusion-catalog-listing 54.1.0", - "datafusion-common 54.1.0", - "datafusion-common-runtime 54.1.0", - "datafusion-datasource 54.1.0", - "datafusion-datasource-arrow 54.1.0", - "datafusion-datasource-csv 54.1.0", - "datafusion-datasource-json 54.1.0", + "datafusion-catalog", + "datafusion-catalog-listing", + "datafusion-common", + "datafusion-common-runtime", + "datafusion-datasource", + "datafusion-datasource-arrow", + "datafusion-datasource-csv", + "datafusion-datasource-json", "datafusion-datasource-parquet", - "datafusion-execution 54.1.0", - "datafusion-expr 54.1.0", - "datafusion-expr-common 54.1.0", - "datafusion-functions 54.1.0", - "datafusion-functions-aggregate 54.1.0", + "datafusion-execution", + "datafusion-expr", + "datafusion-expr-common", + "datafusion-functions", + "datafusion-functions-aggregate", "datafusion-functions-nested", - "datafusion-functions-table 54.1.0", - "datafusion-functions-window 54.1.0", - "datafusion-optimizer 54.1.0", - "datafusion-physical-expr 54.1.0", - "datafusion-physical-expr-adapter 54.1.0", - "datafusion-physical-expr-common 54.1.0", - "datafusion-physical-optimizer 54.1.0", - "datafusion-physical-plan 54.1.0", - "datafusion-session 54.1.0", + "datafusion-functions-table", + "datafusion-functions-window", + "datafusion-optimizer", + "datafusion-physical-expr", + "datafusion-physical-expr-adapter", + "datafusion-physical-expr-common", + "datafusion-physical-optimizer", + "datafusion-physical-plan", + "datafusion-session", "datafusion-sql", "futures", "indexmap 2.14.0", @@ -2612,31 +2566,6 @@ dependencies = [ "uuid", ] -[[package]] -name = "datafusion-catalog" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "37cefde60b26a7f4ff61e9d2ff2833322f91df2b568d7238afe67bde5bdffb66" -dependencies = [ - "arrow", - "async-trait", - "dashmap 6.1.0", - "datafusion-common 53.1.0", - "datafusion-common-runtime 53.1.0", - "datafusion-datasource 53.1.0", - "datafusion-execution 53.1.0", - "datafusion-expr 53.1.0", - "datafusion-physical-expr 53.1.0", - "datafusion-physical-plan 53.1.0", - "datafusion-session 53.1.0", - "futures", - "itertools 0.14.0", - "log", - "object_store", - "parking_lot", - "tokio", -] - [[package]] name = "datafusion-catalog" version = "54.1.0" @@ -2646,14 +2575,14 @@ dependencies = [ "arrow", "async-trait", "dashmap 6.1.0", - "datafusion-common 54.1.0", - "datafusion-common-runtime 54.1.0", - "datafusion-datasource 54.1.0", - "datafusion-execution 54.1.0", - "datafusion-expr 54.1.0", - "datafusion-physical-expr 54.1.0", - "datafusion-physical-plan 54.1.0", - "datafusion-session 54.1.0", + "datafusion-common", + "datafusion-common-runtime", + "datafusion-datasource", + "datafusion-execution", + "datafusion-expr", + "datafusion-physical-expr", + "datafusion-physical-plan", + "datafusion-session", "futures", "itertools 0.14.0", "log", @@ -2662,29 +2591,6 @@ dependencies = [ "tokio", ] -[[package]] -name = "datafusion-catalog-listing" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "17e112307715d6a7a331111a4c2330ff54bc237183511c319e3708a4cff431fb" -dependencies = [ - "arrow", - "async-trait", - "datafusion-catalog 53.1.0", - "datafusion-common 53.1.0", - "datafusion-datasource 53.1.0", - "datafusion-execution 53.1.0", - "datafusion-expr 53.1.0", - "datafusion-physical-expr 53.1.0", - "datafusion-physical-expr-adapter 53.1.0", - "datafusion-physical-expr-common 53.1.0", - "datafusion-physical-plan 53.1.0", - "futures", - "itertools 0.14.0", - "log", - "object_store", -] - [[package]] name = "datafusion-catalog-listing" version = "54.1.0" @@ -2693,43 +2599,21 @@ checksum = "f0668fb32c12065ec242be0e5b4bc62bd7a06a0be3ecd83791ef877e4be67e02" dependencies = [ "arrow", "async-trait", - "datafusion-catalog 54.1.0", - "datafusion-common 54.1.0", - "datafusion-datasource 54.1.0", - "datafusion-execution 54.1.0", - "datafusion-expr 54.1.0", - "datafusion-physical-expr 54.1.0", - "datafusion-physical-expr-adapter 54.1.0", - "datafusion-physical-expr-common 54.1.0", - "datafusion-physical-plan 54.1.0", + "datafusion-catalog", + "datafusion-common", + "datafusion-datasource", + "datafusion-execution", + "datafusion-expr", + "datafusion-physical-expr", + "datafusion-physical-expr-adapter", + "datafusion-physical-expr-common", + "datafusion-physical-plan", "futures", "itertools 0.14.0", "log", "object_store", ] -[[package]] -name = "datafusion-common" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d72a11ca44a95e1081870d3abb80c717496e8a7acb467a1d3e932bb636af5cc2" -dependencies = [ - "ahash", - "arrow", - "arrow-ipc", - "chrono", - "half", - "hashbrown 0.16.1", - "indexmap 2.14.0", - "itertools 0.14.0", - "libc", - "log", - "object_store", - "paste", - "tokio", - "web-time", -] - [[package]] name = "datafusion-common" version = "54.1.0" @@ -2755,17 +2639,6 @@ dependencies = [ "web-time", ] -[[package]] -name = "datafusion-common-runtime" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "89f4afaed29670ec4fd6053643adc749fe3f4bc9d1ce1b8c5679b22c67d12def" -dependencies = [ - "futures", - "log", - "tokio", -] - [[package]] name = "datafusion-common-runtime" version = "54.1.0" @@ -2777,35 +2650,6 @@ dependencies = [ "tokio", ] -[[package]] -name = "datafusion-datasource" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e9fb386e1691355355a96419978a0022b7947b44d4a24a6ea99f00b6b485cbb6" -dependencies = [ - "arrow", - "async-trait", - "bytes", - "chrono", - "datafusion-common 53.1.0", - "datafusion-common-runtime 53.1.0", - "datafusion-execution 53.1.0", - "datafusion-expr 53.1.0", - "datafusion-physical-expr 53.1.0", - "datafusion-physical-expr-adapter 53.1.0", - "datafusion-physical-expr-common 53.1.0", - "datafusion-physical-plan 53.1.0", - "datafusion-session 53.1.0", - "futures", - "glob", - "itertools 0.14.0", - "log", - "object_store", - "rand 0.9.4", - "tokio", - "url", -] - [[package]] name = "datafusion-datasource" version = "54.1.0" @@ -2816,15 +2660,15 @@ dependencies = [ "async-trait", "bytes", "chrono", - "datafusion-common 54.1.0", - "datafusion-common-runtime 54.1.0", - "datafusion-execution 54.1.0", - "datafusion-expr 54.1.0", - "datafusion-physical-expr 54.1.0", - "datafusion-physical-expr-adapter 54.1.0", - "datafusion-physical-expr-common 54.1.0", - "datafusion-physical-plan 54.1.0", - "datafusion-session 54.1.0", + "datafusion-common", + "datafusion-common-runtime", + "datafusion-execution", + "datafusion-expr", + "datafusion-physical-expr", + "datafusion-physical-expr-adapter", + "datafusion-physical-expr-common", + "datafusion-physical-plan", + "datafusion-session", "futures", "glob", "itertools 0.14.0", @@ -2836,30 +2680,6 @@ dependencies = [ "url", ] -[[package]] -name = "datafusion-datasource-arrow" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ffa6c52cfed0734c5f93754d1c0175f558175248bf686c944fb05c373e5fc096" -dependencies = [ - "arrow", - "arrow-ipc", - "async-trait", - "bytes", - "datafusion-common 53.1.0", - "datafusion-common-runtime 53.1.0", - "datafusion-datasource 53.1.0", - "datafusion-execution 53.1.0", - "datafusion-expr 53.1.0", - "datafusion-physical-expr-common 53.1.0", - "datafusion-physical-plan 53.1.0", - "datafusion-session 53.1.0", - "futures", - "itertools 0.14.0", - "object_store", - "tokio", -] - [[package]] name = "datafusion-datasource-arrow" version = "54.1.0" @@ -2870,43 +2690,20 @@ dependencies = [ "arrow-ipc", "async-trait", "bytes", - "datafusion-common 54.1.0", - "datafusion-common-runtime 54.1.0", - "datafusion-datasource 54.1.0", - "datafusion-execution 54.1.0", - "datafusion-expr 54.1.0", - "datafusion-physical-expr-common 54.1.0", - "datafusion-physical-plan 54.1.0", - "datafusion-session 54.1.0", + "datafusion-common", + "datafusion-common-runtime", + "datafusion-datasource", + "datafusion-execution", + "datafusion-expr", + "datafusion-physical-expr-common", + "datafusion-physical-plan", + "datafusion-session", "futures", "itertools 0.14.0", "object_store", "tokio", ] -[[package]] -name = "datafusion-datasource-csv" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "503f29e0582c1fc189578d665ff57d9300da1f80c282777d7eb67bb79fb8cdca" -dependencies = [ - "arrow", - "async-trait", - "bytes", - "datafusion-common 53.1.0", - "datafusion-common-runtime 53.1.0", - "datafusion-datasource 53.1.0", - "datafusion-execution 53.1.0", - "datafusion-expr 53.1.0", - "datafusion-physical-expr-common 53.1.0", - "datafusion-physical-plan 53.1.0", - "datafusion-session 53.1.0", - "futures", - "object_store", - "regex", - "tokio", -] - [[package]] name = "datafusion-datasource-csv" version = "54.1.0" @@ -2916,44 +2713,20 @@ dependencies = [ "arrow", "async-trait", "bytes", - "datafusion-common 54.1.0", - "datafusion-common-runtime 54.1.0", - "datafusion-datasource 54.1.0", - "datafusion-execution 54.1.0", - "datafusion-expr 54.1.0", - "datafusion-physical-expr-common 54.1.0", - "datafusion-physical-plan 54.1.0", - "datafusion-session 54.1.0", + "datafusion-common", + "datafusion-common-runtime", + "datafusion-datasource", + "datafusion-execution", + "datafusion-expr", + "datafusion-physical-expr-common", + "datafusion-physical-plan", + "datafusion-session", "futures", "object_store", "regex", "tokio", ] -[[package]] -name = "datafusion-datasource-json" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e33804749abc8d0c8cb7473228483cb8070e524c6f6086ee1b85a64debe2b3d2" -dependencies = [ - "arrow", - "async-trait", - "bytes", - "datafusion-common 53.1.0", - "datafusion-common-runtime 53.1.0", - "datafusion-datasource 53.1.0", - "datafusion-execution 53.1.0", - "datafusion-expr 53.1.0", - "datafusion-physical-expr-common 53.1.0", - "datafusion-physical-plan 53.1.0", - "datafusion-session 53.1.0", - "futures", - "object_store", - "serde_json", - "tokio", - "tokio-stream", -] - [[package]] name = "datafusion-datasource-json" version = "54.1.0" @@ -2963,14 +2736,14 @@ dependencies = [ "arrow", "async-trait", "bytes", - "datafusion-common 54.1.0", - "datafusion-common-runtime 54.1.0", - "datafusion-datasource 54.1.0", - "datafusion-execution 54.1.0", - "datafusion-expr 54.1.0", - "datafusion-physical-expr-common 54.1.0", - "datafusion-physical-plan 54.1.0", - "datafusion-session 54.1.0", + "datafusion-common", + "datafusion-common-runtime", + "datafusion-datasource", + "datafusion-execution", + "datafusion-expr", + "datafusion-physical-expr-common", + "datafusion-physical-plan", + "datafusion-session", "futures", "object_store", "tokio", @@ -2986,19 +2759,19 @@ dependencies = [ "arrow", "async-trait", "bytes", - "datafusion-common 54.1.0", - "datafusion-common-runtime 54.1.0", - "datafusion-datasource 54.1.0", - "datafusion-execution 54.1.0", - "datafusion-expr 54.1.0", - "datafusion-functions 54.1.0", - "datafusion-functions-aggregate-common 54.1.0", - "datafusion-physical-expr 54.1.0", - "datafusion-physical-expr-adapter 54.1.0", - "datafusion-physical-expr-common 54.1.0", - "datafusion-physical-plan 54.1.0", - "datafusion-pruning 54.1.0", - "datafusion-session 54.1.0", + "datafusion-common", + "datafusion-common-runtime", + "datafusion-datasource", + "datafusion-execution", + "datafusion-expr", + "datafusion-functions", + "datafusion-functions-aggregate-common", + "datafusion-physical-expr", + "datafusion-physical-expr-adapter", + "datafusion-physical-expr-common", + "datafusion-physical-plan", + "datafusion-pruning", + "datafusion-session", "futures", "itertools 0.14.0", "log", @@ -3008,41 +2781,12 @@ dependencies = [ "tokio", ] -[[package]] -name = "datafusion-doc" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8de6ac0df1662b9148ad3c987978b32cbec7c772f199b1d53520c8fa764a87ee" - [[package]] name = "datafusion-doc" version = "54.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d69bb69d8769e34f76839c960dbde24c1ac0c885a79b6c3c2287bdc56ec67891" -[[package]] -name = "datafusion-execution" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c03c7fbdaefcca4ef6ffe425a5fc2325763bfb426599bb0bf4536466efabe709" -dependencies = [ - "arrow", - "arrow-buffer", - "async-trait", - "chrono", - "dashmap 6.1.0", - "datafusion-common 53.1.0", - "datafusion-expr 53.1.0", - "datafusion-physical-expr-common 53.1.0", - "futures", - "log", - "object_store", - "parking_lot", - "rand 0.9.4", - "tempfile", - "url", -] - [[package]] name = "datafusion-execution" version = "54.1.0" @@ -3053,9 +2797,9 @@ dependencies = [ "arrow-buffer", "async-trait", "dashmap 6.1.0", - "datafusion-common 54.1.0", - "datafusion-expr 54.1.0", - "datafusion-physical-expr-common 54.1.0", + "datafusion-common", + "datafusion-expr", + "datafusion-physical-expr-common", "futures", "log", "object_store", @@ -3065,27 +2809,6 @@ dependencies = [ "url", ] -[[package]] -name = "datafusion-expr" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "574b9b6977fedbd2a611cbff12e5caf90f31640ad9dc5870f152836d94bad0dd" -dependencies = [ - "arrow", - "async-trait", - "chrono", - "datafusion-common 53.1.0", - "datafusion-doc 53.1.0", - "datafusion-expr-common 53.1.0", - "datafusion-functions-aggregate-common 53.1.0", - "datafusion-functions-window-common 53.1.0", - "datafusion-physical-expr-common 53.1.0", - "indexmap 2.14.0", - "itertools 0.14.0", - "paste", - "serde_json", -] - [[package]] name = "datafusion-expr" version = "54.1.0" @@ -3096,31 +2819,18 @@ dependencies = [ "arrow-schema", "async-trait", "chrono", - "datafusion-common 54.1.0", - "datafusion-doc 54.1.0", - "datafusion-expr-common 54.1.0", - "datafusion-functions-aggregate-common 54.1.0", - "datafusion-functions-window-common 54.1.0", - "datafusion-physical-expr-common 54.1.0", + "datafusion-common", + "datafusion-doc", + "datafusion-expr-common", + "datafusion-functions-aggregate-common", + "datafusion-functions-window-common", + "datafusion-physical-expr-common", "indexmap 2.14.0", "itertools 0.14.0", "serde_json", "sqlparser", ] -[[package]] -name = "datafusion-expr-common" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7d7c3adf3db8bf61e92eb90cb659c8e8b734593a8f7c8e12a843c7ddba24b87e" -dependencies = [ - "arrow", - "datafusion-common 53.1.0", - "indexmap 2.14.0", - "itertools 0.14.0", - "paste", -] - [[package]] name = "datafusion-expr-common" version = "54.1.0" @@ -3128,39 +2838,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b7b19a8c95522bee8cbb313d74263b85e355d2b52f42e67ef5694bf5de9e9356" dependencies = [ "arrow", - "datafusion-common 54.1.0", + "datafusion-common", "indexmap 2.14.0", "itertools 0.14.0", ] -[[package]] -name = "datafusion-functions" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f28aa4e10384e782774b10e72aca4d93ef7b31aa653095d9d4536b0a3dbc51b6" -dependencies = [ - "arrow", - "arrow-buffer", - "base64 0.22.1", - "chrono", - "chrono-tz 0.10.4", - "datafusion-common 53.1.0", - "datafusion-doc 53.1.0", - "datafusion-execution 53.1.0", - "datafusion-expr 53.1.0", - "datafusion-expr-common 53.1.0", - "datafusion-macros 53.1.0", - "hex", - "itertools 0.14.0", - "log", - "memchr", - "num-traits", - "rand 0.9.4", - "regex", - "unicode-segmentation", - "uuid", -] - [[package]] name = "datafusion-functions" version = "54.1.0" @@ -3174,13 +2856,13 @@ dependencies = [ "blake3", "chrono", "chrono-tz 0.10.4", - "datafusion-common 54.1.0", - "datafusion-doc 54.1.0", - "datafusion-execution 54.1.0", - "datafusion-expr 54.1.0", - "datafusion-expr-common 54.1.0", - "datafusion-macros 54.1.0", - "datafusion-physical-expr-common 54.1.0", + "datafusion-common", + "datafusion-doc", + "datafusion-execution", + "datafusion-expr", + "datafusion-expr-common", + "datafusion-macros", + "datafusion-physical-expr-common", "hex", "itertools 0.14.0", "log", @@ -3193,28 +2875,6 @@ dependencies = [ "uuid", ] -[[package]] -name = "datafusion-functions-aggregate" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "00aa6217e56098ba84e0a338176fe52f0a84cca398021512c6c8c5eff806d0ad" -dependencies = [ - "ahash", - "arrow", - "datafusion-common 53.1.0", - "datafusion-doc 53.1.0", - "datafusion-execution 53.1.0", - "datafusion-expr 53.1.0", - "datafusion-functions-aggregate-common 53.1.0", - "datafusion-macros 53.1.0", - "datafusion-physical-expr 53.1.0", - "datafusion-physical-expr-common 53.1.0", - "half", - "log", - "num-traits", - "paste", -] - [[package]] name = "datafusion-functions-aggregate" version = "54.1.0" @@ -3222,33 +2882,20 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "89bc17041e424a47ed062f43df24d84aab8b57c4c3221e5c1a5eef46d6c5718b" dependencies = [ "arrow", - "datafusion-common 54.1.0", - "datafusion-doc 54.1.0", - "datafusion-execution 54.1.0", - "datafusion-expr 54.1.0", - "datafusion-functions-aggregate-common 54.1.0", - "datafusion-macros 54.1.0", - "datafusion-physical-expr 54.1.0", - "datafusion-physical-expr-common 54.1.0", + "datafusion-common", + "datafusion-doc", + "datafusion-execution", + "datafusion-expr", + "datafusion-functions-aggregate-common", + "datafusion-macros", + "datafusion-physical-expr", + "datafusion-physical-expr-common", "foldhash 0.2.0", "half", "log", "num-traits", ] -[[package]] -name = "datafusion-functions-aggregate-common" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b511250349407db7c43832ab2de63f5557b19a20dfd236b39ca2c04468b50d47" -dependencies = [ - "ahash", - "arrow", - "datafusion-common 53.1.0", - "datafusion-expr-common 53.1.0", - "datafusion-physical-expr-common 53.1.0", -] - [[package]] name = "datafusion-functions-aggregate-common" version = "54.1.0" @@ -3256,9 +2903,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "97dd2a9e865c6108059f5b37b77934f84b50bfb108f837bd0e5c9536e03f0545" dependencies = [ "arrow", - "datafusion-common 54.1.0", - "datafusion-expr-common 54.1.0", - "datafusion-physical-expr-common 54.1.0", + "datafusion-common", + "datafusion-expr-common", + "datafusion-physical-expr-common", ] [[package]] @@ -3269,16 +2916,16 @@ checksum = "75f0bdfeef16d96417b9632ef855645376b242e9006a126dfd0bedfc54a93f5f" dependencies = [ "arrow", "arrow-ord", - "datafusion-common 54.1.0", - "datafusion-doc 54.1.0", - "datafusion-execution 54.1.0", - "datafusion-expr 54.1.0", - "datafusion-expr-common 54.1.0", - "datafusion-functions 54.1.0", - "datafusion-functions-aggregate 54.1.0", - "datafusion-functions-aggregate-common 54.1.0", - "datafusion-macros 54.1.0", - "datafusion-physical-expr-common 54.1.0", + "datafusion-common", + "datafusion-doc", + "datafusion-execution", + "datafusion-expr", + "datafusion-expr-common", + "datafusion-functions", + "datafusion-functions-aggregate", + "datafusion-functions-aggregate-common", + "datafusion-macros", + "datafusion-physical-expr-common", "hashbrown 0.17.1", "itertools 0.14.0", "itoa", @@ -3286,22 +2933,6 @@ dependencies = [ "memchr", ] -[[package]] -name = "datafusion-functions-table" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "72b40d3f5bbb3905f9ccb1ce9485a9595c77b69758a7c24d3ba79e334ff51e7e" -dependencies = [ - "arrow", - "async-trait", - "datafusion-catalog 53.1.0", - "datafusion-common 53.1.0", - "datafusion-expr 53.1.0", - "datafusion-physical-plan 53.1.0", - "parking_lot", - "paste", -] - [[package]] name = "datafusion-functions-table" version = "54.1.0" @@ -3310,32 +2941,14 @@ checksum = "f4e4941673c917819616877e9993da4503e4f4739812be0bc32c5356184c6383" dependencies = [ "arrow", "async-trait", - "datafusion-catalog 54.1.0", - "datafusion-common 54.1.0", - "datafusion-expr 54.1.0", - "datafusion-physical-expr 54.1.0", - "datafusion-physical-plan 54.1.0", + "datafusion-catalog", + "datafusion-common", + "datafusion-expr", + "datafusion-physical-expr", + "datafusion-physical-plan", "parking_lot", ] -[[package]] -name = "datafusion-functions-window" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d4e88ec9d57c9b685d02f58bfee7be62d72610430ddcedb82a08e5d9925dbfb6" -dependencies = [ - "arrow", - "datafusion-common 53.1.0", - "datafusion-doc 53.1.0", - "datafusion-expr 53.1.0", - "datafusion-functions-window-common 53.1.0", - "datafusion-macros 53.1.0", - "datafusion-physical-expr 53.1.0", - "datafusion-physical-expr-common 53.1.0", - "log", - "paste", -] - [[package]] name = "datafusion-functions-window" version = "54.1.0" @@ -3343,45 +2956,24 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "12dd2e16c12b84b6f6b41b19f55b366dd1c46876bb35b86896c6349067379e8d" dependencies = [ "arrow", - "datafusion-common 54.1.0", - "datafusion-doc 54.1.0", - "datafusion-expr 54.1.0", - "datafusion-functions-window-common 54.1.0", - "datafusion-macros 54.1.0", - "datafusion-physical-expr 54.1.0", - "datafusion-physical-expr-common 54.1.0", + "datafusion-common", + "datafusion-doc", + "datafusion-expr", + "datafusion-functions-window-common", + "datafusion-macros", + "datafusion-physical-expr", + "datafusion-physical-expr-common", "log", ] -[[package]] -name = "datafusion-functions-window-common" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8307bb93519b1a91913723a1130cfafeee3f72200d870d88e91a6fc5470ede5c" -dependencies = [ - "datafusion-common 53.1.0", - "datafusion-physical-expr-common 53.1.0", -] - [[package]] name = "datafusion-functions-window-common" version = "54.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4cdc5e4b6f8b6ef823cc1c761f85088ad4c884fe8df64df3cbcc6b2b84698441" dependencies = [ - "datafusion-common 54.1.0", - "datafusion-physical-expr-common 54.1.0", -] - -[[package]] -name = "datafusion-macros" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2e367e6a71051d0ebdd29b2f85d12059b38b1d1f172c6906e80016da662226bd" -dependencies = [ - "datafusion-doc 53.1.0", - "quote", - "syn 2.0.117", + "datafusion-common", + "datafusion-physical-expr-common", ] [[package]] @@ -3390,30 +2982,11 @@ version = "54.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1a3614234dd93578c92428cb4f408e020874f0d2b7e6c90c928d9d28b5df2ceb" dependencies = [ - "datafusion-doc 54.1.0", + "datafusion-doc", "quote", "syn 2.0.117", ] -[[package]] -name = "datafusion-optimizer" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e929015451a67f77d9d8b727b2bf3a40c4445fdef6cdc53281d7d97c76888ace" -dependencies = [ - "arrow", - "chrono", - "datafusion-common 53.1.0", - "datafusion-expr 53.1.0", - "datafusion-expr-common 53.1.0", - "datafusion-physical-expr 53.1.0", - "indexmap 2.14.0", - "itertools 0.14.0", - "log", - "regex", - "regex-syntax", -] - [[package]] name = "datafusion-optimizer" version = "54.1.0" @@ -3422,10 +2995,10 @@ checksum = "a0635620b050b81bb92764e99250868f654e2cd5ad1bece413283b3f73c83179" dependencies = [ "arrow", "chrono", - "datafusion-common 54.1.0", - "datafusion-expr 54.1.0", - "datafusion-expr-common 54.1.0", - "datafusion-physical-expr 54.1.0", + "datafusion-common", + "datafusion-expr", + "datafusion-expr-common", + "datafusion-physical-expr", "indexmap 2.14.0", "itertools 0.14.0", "log", @@ -3433,29 +3006,6 @@ dependencies = [ "regex-syntax", ] -[[package]] -name = "datafusion-physical-expr" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4b1e68aba7a4b350401cfdf25a3d6f989ad898a7410164afe9ca52080244cb59" -dependencies = [ - "ahash", - "arrow", - "datafusion-common 53.1.0", - "datafusion-expr 53.1.0", - "datafusion-expr-common 53.1.0", - "datafusion-functions-aggregate-common 53.1.0", - "datafusion-physical-expr-common 53.1.0", - "half", - "hashbrown 0.16.1", - "indexmap 2.14.0", - "itertools 0.14.0", - "parking_lot", - "paste", - "petgraph", - "tokio", -] - [[package]] name = "datafusion-physical-expr" version = "54.1.0" @@ -3463,11 +3013,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8cabf7a86eb70b816729e33c81bf7767c936ee1226f607a114f5dac2decac8d0" dependencies = [ "arrow", - "datafusion-common 54.1.0", - "datafusion-expr 54.1.0", - "datafusion-expr-common 54.1.0", - "datafusion-functions-aggregate-common 54.1.0", - "datafusion-physical-expr-common 54.1.0", + "datafusion-common", + "datafusion-expr", + "datafusion-expr-common", + "datafusion-functions-aggregate-common", + "datafusion-physical-expr-common", "half", "hashbrown 0.17.1", "indexmap 2.14.0", @@ -3477,21 +3027,6 @@ dependencies = [ "tokio", ] -[[package]] -name = "datafusion-physical-expr-adapter" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ea22315f33cf2e0adc104e8ec42e285f6ed93998d565c65e82fec6a9ee9f9db4" -dependencies = [ - "arrow", - "datafusion-common 53.1.0", - "datafusion-expr 53.1.0", - "datafusion-functions 53.1.0", - "datafusion-physical-expr 53.1.0", - "datafusion-physical-expr-common 53.1.0", - "itertools 0.14.0", -] - [[package]] name = "datafusion-physical-expr-adapter" version = "54.1.0" @@ -3499,29 +3034,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "de222e04f7e6744501555a54ab0abe26bfdfebee380af79a9bdc175704246859" dependencies = [ "arrow", - "datafusion-common 54.1.0", - "datafusion-expr 54.1.0", - "datafusion-functions 54.1.0", - "datafusion-physical-expr 54.1.0", - "datafusion-physical-expr-common 54.1.0", - "itertools 0.14.0", -] - -[[package]] -name = "datafusion-physical-expr-common" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b04b45ea8ad3ac2d78f2ea2a76053e06591c9629c7a603eda16c10649ecf4362" -dependencies = [ - "ahash", - "arrow", - "chrono", - "datafusion-common 53.1.0", - "datafusion-expr-common 53.1.0", - "hashbrown 0.16.1", - "indexmap 2.14.0", + "datafusion-common", + "datafusion-expr", + "datafusion-functions", + "datafusion-physical-expr", + "datafusion-physical-expr-common", "itertools 0.14.0", - "parking_lot", ] [[package]] @@ -3532,8 +3050,8 @@ checksum = "72d0d0057fc5a502d45c870cb6d47c66eb7bdd5edb1bd71ad6f3f724975ac2a8" dependencies = [ "arrow", "chrono", - "datafusion-common 54.1.0", - "datafusion-expr-common 54.1.0", + "datafusion-common", + "datafusion-expr-common", "hashbrown 0.17.1", "indexmap 2.14.0", "itertools 0.14.0", @@ -3541,24 +3059,6 @@ dependencies = [ "pin-project", ] -[[package]] -name = "datafusion-physical-optimizer" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7cb13397809a425918f608dfe8653f332015a3e330004ab191b4404187238b95" -dependencies = [ - "arrow", - "datafusion-common 53.1.0", - "datafusion-execution 53.1.0", - "datafusion-expr 53.1.0", - "datafusion-expr-common 53.1.0", - "datafusion-physical-expr 53.1.0", - "datafusion-physical-expr-common 53.1.0", - "datafusion-physical-plan 53.1.0", - "datafusion-pruning 53.1.0", - "itertools 0.14.0", -] - [[package]] name = "datafusion-physical-optimizer" version = "54.1.0" @@ -3566,49 +3066,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "86046eed10950c5f9aaed9acfd148e9bd2e1dfdfe4f9aef607d1447b271e4183" dependencies = [ "arrow", - "datafusion-common 54.1.0", - "datafusion-execution 54.1.0", - "datafusion-expr 54.1.0", - "datafusion-expr-common 54.1.0", - "datafusion-physical-expr 54.1.0", - "datafusion-physical-expr-common 54.1.0", - "datafusion-physical-plan 54.1.0", - "datafusion-pruning 54.1.0", + "datafusion-common", + "datafusion-execution", + "datafusion-expr", + "datafusion-expr-common", + "datafusion-physical-expr", + "datafusion-physical-expr-common", + "datafusion-physical-plan", + "datafusion-pruning", "itertools 0.14.0", ] -[[package]] -name = "datafusion-physical-plan" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5edc023675791af9d5fb4cc4c24abf5f7bd3bd4dcf9e5bd90ea1eff6976dcc79" -dependencies = [ - "ahash", - "arrow", - "arrow-ord", - "arrow-schema", - "async-trait", - "datafusion-common 53.1.0", - "datafusion-common-runtime 53.1.0", - "datafusion-execution 53.1.0", - "datafusion-expr 53.1.0", - "datafusion-functions 53.1.0", - "datafusion-functions-aggregate-common 53.1.0", - "datafusion-functions-window-common 53.1.0", - "datafusion-physical-expr 53.1.0", - "datafusion-physical-expr-common 53.1.0", - "futures", - "half", - "hashbrown 0.16.1", - "indexmap 2.14.0", - "itertools 0.14.0", - "log", - "num-traits", - "parking_lot", - "pin-project-lite", - "tokio", -] - [[package]] name = "datafusion-physical-plan" version = "54.1.0" @@ -3621,15 +3089,15 @@ dependencies = [ "arrow-ord", "arrow-schema", "async-trait", - "datafusion-common 54.1.0", - "datafusion-common-runtime 54.1.0", - "datafusion-execution 54.1.0", - "datafusion-expr 54.1.0", - "datafusion-functions 54.1.0", - "datafusion-functions-aggregate-common 54.1.0", - "datafusion-functions-window-common 54.1.0", - "datafusion-physical-expr 54.1.0", - "datafusion-physical-expr-common 54.1.0", + "datafusion-common", + "datafusion-common-runtime", + "datafusion-execution", + "datafusion-expr", + "datafusion-functions", + "datafusion-functions-aggregate-common", + "datafusion-functions-window-common", + "datafusion-physical-expr", + "datafusion-physical-expr-common", "futures", "half", "hashbrown 0.17.1", @@ -3642,23 +3110,6 @@ dependencies = [ "tokio", ] -[[package]] -name = "datafusion-pruning" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ac8c76860e355616555081cab5968cec1af7a80701ff374510860bcd567e365a" -dependencies = [ - "arrow", - "datafusion-common 53.1.0", - "datafusion-datasource 53.1.0", - "datafusion-expr-common 53.1.0", - "datafusion-physical-expr 53.1.0", - "datafusion-physical-expr-common 53.1.0", - "datafusion-physical-plan 53.1.0", - "itertools 0.14.0", - "log", -] - [[package]] name = "datafusion-pruning" version = "54.1.0" @@ -3666,29 +3117,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "eb63eeac6de19be40f487b65dd84e546195f783c5a9928618e0c4f2a3569b0d7" dependencies = [ "arrow", - "datafusion-common 54.1.0", - "datafusion-datasource 54.1.0", - "datafusion-expr-common 54.1.0", - "datafusion-physical-expr 54.1.0", - "datafusion-physical-expr-common 54.1.0", - "datafusion-physical-plan 54.1.0", + "datafusion-common", + "datafusion-datasource", + "datafusion-expr-common", + "datafusion-physical-expr", + "datafusion-physical-expr-common", + "datafusion-physical-plan", "log", ] -[[package]] -name = "datafusion-session" -version = "53.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5412111aa48e2424ba926112e192f7a6b7e4ccb450145d25ce5ede9f19dc491e" -dependencies = [ - "async-trait", - "datafusion-common 53.1.0", - "datafusion-execution 53.1.0", - "datafusion-expr 53.1.0", - "datafusion-physical-plan 53.1.0", - "parking_lot", -] - [[package]] name = "datafusion-session" version = "54.1.0" @@ -3696,10 +3133,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5f961d209177f91bd014db5cbb2c33b7d28a2597b9003e77f17aeb712964315a" dependencies = [ "async-trait", - "datafusion-common 54.1.0", - "datafusion-execution 54.1.0", - "datafusion-expr 54.1.0", - "datafusion-physical-plan 54.1.0", + "datafusion-common", + "datafusion-execution", + "datafusion-expr", + "datafusion-physical-plan", "parking_lot", ] @@ -3712,8 +3149,8 @@ dependencies = [ "arrow", "bigdecimal", "chrono", - "datafusion-common 54.1.0", - "datafusion-expr 54.1.0", + "datafusion-common", + "datafusion-expr", "datafusion-functions-nested", "indexmap 2.14.0", "log", @@ -4653,7 +4090,7 @@ dependencies = [ "arrow-array", "arrow-buffer", "arrow-schema", - "datafusion 54.1.0", + "datafusion", "geo", "geo-traits", "geoarrow-array", @@ -6032,11 +5469,11 @@ dependencies = [ "crossbeam-queue", "crossbeam-skiplist", "dashmap 6.1.0", - "datafusion 54.1.0", - "datafusion-expr 54.1.0", - "datafusion-functions 54.1.0", - "datafusion-physical-expr 54.1.0", - "datafusion-physical-plan 54.1.0", + "datafusion", + "datafusion-expr", + "datafusion-functions", + "datafusion-physical-expr", + "datafusion-physical-plan", "either", "fst", "futures", @@ -6155,7 +5592,7 @@ dependencies = [ "async-trait", "byteorder", "bytes", - "datafusion-common 54.1.0", + "datafusion-common", "datafusion-sql", "futures", "itertools 0.14.0", @@ -6197,10 +5634,10 @@ dependencies = [ "arrow-select", "async-trait", "chrono", - "datafusion 54.1.0", - "datafusion-common 54.1.0", - "datafusion-functions 54.1.0", - "datafusion-physical-expr 54.1.0", + "datafusion", + "datafusion-common", + "datafusion-functions", + "datafusion-physical-expr", "futures", "jsonb", "lance-arrow", @@ -6298,7 +5735,7 @@ dependencies = [ "async-trait", "byteorder", "bytes", - "datafusion-common 54.1.0", + "datafusion-common", "futures", "lance-arrow", "lance-core", @@ -6320,7 +5757,7 @@ version = "9.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "97739c10f2c818ac4aa8a798a115e11c790c174d016c0836792aa3b8d0f981ab" dependencies = [ - "datafusion 54.1.0", + "datafusion", "geo-traits", "geo-types", "geoarrow-array", @@ -6340,10 +5777,10 @@ dependencies = [ "async-trait", "bytes", "chrono", - "datafusion 54.1.0", - "datafusion-common 54.1.0", - "datafusion-expr 54.1.0", - "datafusion-functions-aggregate 54.1.0", + "datafusion", + "datafusion-common", + "datafusion-expr", + "datafusion-functions-aggregate", "datafusion-sql", "futures", "lance", @@ -6365,7 +5802,7 @@ version = "0.5.4" dependencies = [ "arrow-schema", "async-trait", - "datafusion 54.1.0", + "datafusion", "lance-namespace", "reqwest 0.12.28", "serde", @@ -6447,10 +5884,10 @@ dependencies = [ "bytes", "chrono", "crossbeam-queue", - "datafusion 54.1.0", - "datafusion-common 54.1.0", - "datafusion-expr 54.1.0", - "datafusion-physical-expr 54.1.0", + "datafusion", + "datafusion-common", + "datafusion-expr", + "datafusion-physical-expr", "dirs", "fst", "futures", @@ -6509,9 +5946,9 @@ dependencies = [ "arrow-select", "async-trait", "bytes", - "datafusion 54.1.0", - "datafusion-common 54.1.0", - "datafusion-expr 54.1.0", + "datafusion", + "datafusion-common", + "datafusion-expr", "futures", "lance-core", "lance-io", @@ -7314,7 +6751,6 @@ version = "0.1.0" dependencies = [ "arrow", "bgz17", - "datafusion 53.1.0", "graph-flow", "lance-graph", "lance-graph-contract", diff --git a/claude-notes/plans/2026-08-15-osm-lance-lifecycle.md b/claude-notes/plans/2026-08-15-osm-lance-lifecycle.md new file mode 100644 index 000000000..2901c2b92 --- /dev/null +++ b/claude-notes/plans/2026-08-15-osm-lance-lifecycle.md @@ -0,0 +1,427 @@ +# OSM/Lance lifecycle — three clocks, one lineage per region + +**Status:** Phase A implemented, corrected once, tests re-run. Phases B–F not +started (Phase B observational surface not begun this session either). +**Operator directive:** 2026-08-15, this session. +**Upstream contracts:** `AdaWorldAPI/lance-graph` PRs **#912** (baseline) and +**#913** (the five semantic gates). #911 is historical context and is +superseded wherever the contracts differ. +**Correction pass:** 2026-08-15, same session — nine structural corrections to +the first Phase-A draft (product-state lifecycle, validated identity types, +retention authority gating, head-behind handling, contract-probe labeling). +Recorded below in place; see § Correction pass for the full list. This is a +correction **within** the three-clock/one-lineage architecture, not a +redesign of it. + +## The correction this plan exists to encode + +A live Lance dataset is a **versioned lineage, not an immutable directory**. A +legitimate append, index build, compaction, reopen, or atime change must not +invalidate the imported OSM source. So: + +- **Never** hash the whole Lance directory on every boot. +- **Never** compare its changing object inventory to one static checksum. + +The current code does the moral equivalent — a full SHA-256 of every artifact +on every boot (even on a warm cache hit) plus a second full pass to hash the +slab for the dataset-freshness check. That is the measured cause of "memory +climbs back to ~4 GB after a restart with no map traffic". + +## The three clocks — never conflated + +| # | Clock | Fields | +|---|---|---| +| 1 | **Semantic import identity** | `source_bake_id`, `source_manifest_sha256`, `import_id`, `origin_seal`, `import_seal` | +| 2 | **Physical Lance history** | `publication_version`, `serving_version`, `observed_head` | +| 3 | **Replica progress** | storage root (volume/ephemeral), `mirrored_through_version`, S3 lag | + +The failure mode this prevents, in lance-graph's own words (#913): retrying +cycle 1 while the head stood at V5 recorded cycle 1 as *"sealed into V5"*. A +later observed head is a **read horizon**, not evidence that the origin changed. + +## Origin seal — internally non-contradictory (correction) + +The first draft stored `OriginSeal.lineage: LineageId` (which already carries +`layout`) **and** a separate, independently-settable `OriginSeal.layout` — two +copies of the same fact with no constraint tying them together, so a +hand-built seal could disagree with itself. It also had no field naming the +import that *created* the lineage, which means the first sealed import was +unfindable from the seal alone. + +Fixed by construction: `OriginSeal` fields are private; the only constructor +is `OriginSeal::create(identity: ImportIdentity, origin_publication)`. `layout` +is now an accessor that reads `identity.lineage.layout` — there is no second +storage location to drift. `origin_import_id` is *derived* +(`identity.import_id()`) and exposed as an accessor, so it can never disagree +with the identity that produced it, and it is now what makes the origin +import findable. + +## Publication semantics (mirrored from lance-graph #913) + +- Fresh successful import → `publication_version = Some(actual_version)`. +- Reconciled retry → `publication_version = None`, `observed_head = Some(head)`. +- Never assign a reconciled import's current head as its historical publication + position. +- Recover an unknown historical position by **searching the sealed `import_id` + in Lance history**, never by inference. + +## One lineage per region + +``` +berlin/osm-lance-v2 +brandenburg/osm-lance-v2 +``` + +A new SOA bake inside `osm-lance-v2` is a **new sealed import version in the +same lineage**. Only an incompatible layout mints `osm-lance-v3`. No new S3 +dataset prefix per import. + +**Correction: the region is a validated `RegionId`, not an arbitrary +`String`.** It is interpolated into an S3 key and joined onto a filesystem +path, so `..`, `/`, and `\` would let a stray environment variable read a +different prefix or write outside the artifact root. `RegionId::parse` +enforces the same rule the existing slab hydrator already applies — +non-empty, `<= 64` bytes, lowercase `[a-z0-9-]` — as a *type*, so the +guarantee travels with the value instead of being re-checked (or forgotten) +at each new use site. `SourceBakeId` gets the equivalent treatment +(`[a-z0-9_.-]`, no `..`) since it is the kind of value that grows into staging +directory names. + +## S3 key separation + +``` +.config/q2/osm/config.yaml desired state +q2/lance/// the native lineage +q2/lance///origin-seal.yaml +q2/lance///import-seals/.yaml +q2/lance///published.yaml pointer, written LAST +.state/q2/osm///status.yaml +.state/q2/osm///last-error.yaml +.runs/q2/osm////part-*.jsonl +``` + +Desired config carries **no mutable `imported: true/false`**. Completion is a +valid sealed import identity, not a flag. + +## Hash choices — deliberate, not incidental + +- **FNV-1a/64** stays the fast physical batch marker (`import_id` / batch hash), + matching the existing Phase-A contract. Not a security property; an + idempotency key. +- **SHA-256** stays for externally published SOA manifests and transport + verification, held as **parsed 32 bytes** (`SourceManifestSha256`), not a + string — so `AB12…` and `ab12…` derive the *same* import identity instead of + two, which a text-hashed field would silently allow. +- **No BLAKE3 in this work.** +- **Correction: the marker narrows, the tuple decides.** Because FNV-1a/64 is + 64-bit and non-cryptographic, a marker match is a *candidate*, not a proof. + `reconcile(candidate, found)` compares the full identity tuple + `(lineage, bake, source_manifest_sha256, layout_abi, writer_abi)` after the + marker matches, and returns `SameImport` / `HashConflict` / + `DifferentImport`. A same-marker-different-tuple result is a **permanent** + conflict (mirrors `lance-graph`'s `CommitError::HashConflict`) — reconciling + it would silently adopt another import's data as this one's. + +## Storage policy (ordered) + +1. `OSM_ARTIFACT_ROOT` +2. `OSM_SLAB_CACHE_DIR` (existing) +3. `RAILWAY_VOL/osm` +4. `/volume01/osm` when present and writable +5. documented ephemeral fallback + +The selected root records **`volume` or `ephemeral`** — that kind drives the +activation gate below. + +**Correction: a named path is not a durable path.** The first draft classified +candidates 1 and 2 as `Volume` unconditionally, on the reasoning that "an +operator who names a root is asserting durability." That reasoning does not +hold: `OSM_ARTIFACT_ROOT` and `OSM_SLAB_CACHE_DIR` are generic overrides that +routinely point at container-local scratch, and misclassifying one as +`Volume` would let an ephemeral candidate activate without S3 — silently +reintroducing the exact cost (re-running the whole SOA→Lance conversion on +every redeploy) this design exists to remove. Only `RAILWAY_VOL` and a +probed-writable `/volume01` are *intrinsically* known to be volume storage. +Candidates 1 and 2 now take their kind from an explicit +`RootInputs::override_durability: Option`, defaulting to +`Ephemeral` when the operator hasn't said — the fail-safe direction, since +being wrongly cautious costs one S3 wait and being wrongly confident costs a +full re-import. Also fixed in the same pass: `RootInputs::default()` — which +`Default::derive` would leave as an empty `PathBuf` resolving against the +process's current directory — now resolves the ephemeral fallback to +`std::env::temp_dir()`, and `resolve_root` refuses a blank *or relative* +override path (both mean "different processes, different places"), falling +through to the next candidate instead. + +## Activation gate + +```yaml +activation: + volume: + require_s3_before_activation: false + ephemeral: + require_s3_before_activation: true +``` + +A persistent volume may activate a locally verified candidate while S3 catches +up, reporting `s3_lagging`. An **ephemeral** candidate is not fully successful +until S3 publication succeeds — otherwise the next redeploy repeats the whole +SOA→Lance conversion, which is the cost this design exists to remove. + +## Update A → B, ordered + +1. Build B in unique local staging. +2. **Reconcile `import_id` before creating another version.** +3. Commit B through one owned writer. +4. Validate B. +5. Ensure A is durable on S3 as `previous_clean`. +6. Upload only Lance objects **absent** from the S3 lineage. +7. Publish the S3 head pointer **last**, with generation/ETag protection. +8. Atomically activate B locally. +9. Mark A `previous_clean`. +10. Release Arrow buffers, writer/session state, source mappings, clean + page-cache ranges. +11. GC older unreferenced versions only after a grace period. + +**There is no compensating delete after an ambiguous publication.** Re-submit +the same frozen import identity and reconcile first. Orphaned unreferenced S3 +objects are harmless and collected later. + +## Retention + +Keep active clean, previous clean, and **all** small seals. Never retain a full +independent dataset directory per import. Never upload a fragment already +referenced in S3. + +**Correction: "keep" is a reachability closure, and destructive GC is +deferred, not merely gated by a grace period.** The first draft modeled +retention as `everything - keep` once a grace period elapsed. Two problems: + +1. `keep` must be **everything reachable from a pinned Lance version** — + manifest, fragments, indices, **deletion files**, transaction records, + required metadata — not "the fragments I happened to upload." A partial + inventory differenced against "everything" deletes exactly the objects it + failed to enumerate, which is worse than not collecting at all. +2. No mechanism in Phase A–C can actually *prove* an object is unreachable + from every retained version. `plan_retention` now takes a + [`GcAuthority`]: `None` (Phases A–C — nothing may be deleted, full stop) + or `LanceNativeReachability` (Phase D). `RetentionPlan` is + `KeepOnly { keep, hold }` until all three gates open together — complete + inventories, elapsed grace period, and `GcAuthority` — at which point it + becomes `Collectable { keep, collect }`. **An incomplete inventory holds + regardless of the other two gates**, tested directly. + +Phase A therefore *specifies retention intent* (what must be kept, and why) +and produces **zero deletion plans**, by construction, until Phase D ships the +real reachability mechanism. That is a stronger statement than "the grace +period holds" — it is "there is no code path in Phase A–C that can return a +non-empty `collect` set." + +**Honest limit, unchanged by the correction:** while the importer performs a +full `Overwrite`, active + previous clean still costs ~**two complete map +datasets**. Version metadata cannot deduplicate two entirely rewritten +fragment sets. Fixing that needs the Phase-E native-columnar Morton-shard +update — it is not a retention-policy problem and this plan does not pretend +otherwise. + +## Serving state and import state are orthogonal (correction) + +The first draft made `LifecycleState` a single sum type where +`ImportStarted` replaced `Active` outright — so a candidate import's failure +had nothing to restore *to*: the artifact being served had been absorbed into +the import's own state and vanished with it. Caught by +`a_failed_candidate_never_disturbs_the_active_artifact` failing. + +The fix is a **product**, not a smarter sum: + +- `Lifecycle.serving: Option` — what this + replica serves, full stop. +- `Lifecycle.operation: Operation::{Idle, Importing{candidate, generation}, + Failed{candidate, generation, error}}` — what an import is doing, orthogonal + to serving. + +`activate()`/`fail()` take `&self` and return `Result` +— they never mutate `serving` except `activate()`'s own rotation, which is the +only place rotation happens. Both are **fenced by generation**: starting +candidate C advances a monotonic counter, and a late completion for a +superseded candidate B is rejected as `LifecycleError::StaleCompletion` +rather than activating or failing C. This is the concurrency case the earlier +draft had no way to express at all. + +## Serving model + +Replace the env-var/`OnceLock` activation with an `OsmArtifactManager` in +`AppState` holding an atomically replaceable `Arc` to the active dataset. +In-flight readers finish against the old `Arc`; new readers get the new one. +**The listener and global `/health` must not wait for hydration.** + +`/api/osm/status` reports lifecycle state, lineage + origin seal, desired / +active / previous-clean import identities, publication / serving / observed +versions, replica state, S3 lag, import progress, seal verification, last +classified error, and the cgroup memory split — **without touching map data +pages**. + +A checksum mismatch in an obsolete whole-directory inventory is **not** +corruption. A missing or unreadable object actually referenced by the selected +sealed version **is** a candidate failure → fall back to `previous_clean`; +never discard unrelated valid history. + +## Head drift, and head-BEHIND (correction) + +If the observed head is later than the known clean import: + +- recognise known maintenance descendants when possible — and **say whether + that recognition is verified or merely asserted** (`MaintenanceProvenance`; + Phase A has no history walker, so it is always `CallerAsserted` today, named + as a limitation rather than hidden); +- otherwise serve a **checkout of the last known clean sealed version**; +- report `head_drift`; +- never discard the lineage merely because the head advanced. + +**A head *behind* the clean import is a different failure and was folded into +"drift" in the first draft — that conflation is wrong.** `observed_head < +clean` is not an unexplained later head; it means the local replica is +truncated or stale, or the requested clean version simply is not present here. +`decide_head` now returns `HeadDecision::ReplicaIncomplete { reason: +HeadBehindClean | CleanVersionAbsent }` for both cases, and — critically — +does **not** offer a checkout of a version the replica may not possess. +`recover_behind` decides separately: hydrate the clean version from S3 if it +has it, else serve the newest available clean version strictly below the +target (a real degradation, chosen only when S3 cannot help), else report +`Unserviceable`. Nothing is deleted and the lineage is not discarded in any of +these paths. + +## Phases (PR-sized) + +- **A.** Pure lifecycle types, state machine, manifests, storage resolution, + falsifiers. ← *this session* +- **B.** `OsmArtifactManager`, fast listener startup, status endpoint, memory + telemetry. ← *observational parts only, if cleanly separable* +- **C.** Versioned local import with origin/import seals and reconciliation. +- **D.** Incremental S3 lineage mirroring, publication pointer, recovery, + retention. +- **E.** Native Lance column layout + Morton-shard updates; removes the single + `FixedSizeBinary`/raw-offset contract and the Brandenburg row ceiling. +- **F.** UI status/banner wiring through q2 / a2ui. + +## Falsifiers — contract probes (Phase A, shipped) vs. system falsifiers (still open) + +**Correction: every item below was a decision-function test, and Phase A has +no running system for a decision to be *wrong about*.** The first draft +listed these as flatly "written before implementation" / done, without +distinguishing "this proves the decision function returns the right answer" +from "this proves the real boot/import/publish path obeys it." They are not +the same claim, and conflating them is how a green suite comes to certify +behaviour nobody measured. Each row below now names both halves; the second +column is **owed, not shipped**, and stays open until the phase that can +actually observe the real system lands it. + +| Property | Phase-A contract probe | System falsifier (owed) | +|---|---|---| +| zero full-data traversal on warm boot | [x] `the_warm_boot_read_plan_names_only_small_seals` — the *plan* names no data object | [ ] Phase B/C: instrument the real boot path, assert bytes read are bounded by seal size | +| access residue does not change semantic identity | [x] `access_residue_does_not_change_semantic_identity` | [ ] Phase C: run a real import, compact the dataset, assert the seal on disk is unchanged | +| observed head later than publication | [x] `a_reconciled_retry_records_no_publication_version` | [ ] Phase C: assert against a real `Dataset::versions()` | +| reconciled retry creates no second version | [x] constructing `Reconciled` applies the rule | [ ] Phase C: assert `Dataset::versions()` gained no entry across a real reconciled retry | +| successful active→previous rotation | [x] `activation_rotates_active_to_previous` | [ ] Phase C: real activation, real filesystem/S3 state | +| failed candidate preserves active AND previous | [x] `a_failed_candidate_never_disturbs_active_or_previous` | [ ] Phase C: real failure path | +| stale completion cannot activate/fail a superseding candidate | [x] `a_stale_completion_cannot_activate_or_fail_the_superseding_candidate` (new — the concurrency case the first draft could not express) | [ ] Phase C: real concurrent imports | +| pointer-last S3 publication | [x] `the_pointer_is_published_last_and_present_objects_are_not_re_uploaded` | [ ] Phase D: observe real S3 write order under a killed process | +| ephemeral S3 activation gate | [x] `an_ephemeral_candidate_waits_for_s3_and_a_volume_does_not` | [ ] Phase D: real S3 state | +| generic override is ephemeral unless durability is declared | [x] `a_generic_override_is_ephemeral_unless_durability_is_declared` (new) | [ ] Phase B: real env-var wiring | +| retention keeps active/previous/seals; authorises no deletion in A–C | [x] `retention_keeps_everything_reachable_and_authorises_no_deletion_in_phase_a` | [ ] Phase D: real Lance-native reachability mechanism | +| incomplete inventory ⇒ no deletion plan, ever | [x] `an_incomplete_inventory_produces_no_deletion_plan` (new) | [ ] Phase D | +| bounded active+previous storage after cleanup | [x] proven inside the retention test above | [ ] Phase D | +| unknown-head checkout of last clean | [x] `an_unexplained_later_head_serves_the_clean_checkout_and_reports_drift` | [ ] Phase C: real Lance history | +| head behind clean is a replica problem, not drift (new) | [x] `a_head_behind_clean_is_a_replica_problem_not_a_drift`, `an_absent_clean_version_is_never_offered_as_a_checkout` | [ ] Phase D: real recovery path | +| recovery hydrates before degrading (new) | [x] `recovery_hydrates_when_it_can_and_degrades_only_when_it_must` | [ ] Phase D | +| layout incompatibility creates a new lineage | [x] `an_incompatible_layout_is_a_different_lineage` | — (pure identity property, no system half owed) | +| first-create ambiguity | [x] `the_first_activation_has_no_previous`; `an_ambiguous_publication_seals_nothing` proves `Ambiguous ⇒ None` | [ ] Phase C/D: kill the process between commit and ack on a first create, restart, re-submit, assert exactly one version — `Ambiguous ⇒ None` is NOT this falsifier by itself | +| telemetry failure does not affect serving | — not attempted this session | [ ] Phase B: no telemetry sink exists yet; open | +| a marker match is a candidate, the tuple decides (new) | [x] `a_marker_match_with_a_different_tuple_is_a_conflict_not_a_reconcile` | [ ] Phase C: a real FNV collision (or a forced one) against real history | +| digest casing cannot change identity (new) | [x] `digest_casing_cannot_change_the_import_identity` | — (pure identity property) | +| region/bake id reject path traversal (new) | [x] `region_rejects_everything_that_could_escape_a_prefix_or_a_directory`, `bake_id_rejects_traversal_and_separators` | [ ] Phase C/D: a real S3 key built from a rejected value must never be attempted | +| origin seal is internally non-contradictory (new) | [x] `an_origin_seal_is_internally_consistent_and_names_its_import` | — (pure identity property) | +| default root is never the current directory (new) | [x] `the_default_root_is_absolute_and_never_the_current_directory` | — (pure identity property) | + +## Work items + +### Phase A +- [x] `RegionId`, `SourceBakeId`, `SourceManifestSha256` — validated identity + primitives (path-traversal-safe, casing-invariant) +- [x] `LineageId`, `ImportIdentity`, `OriginSeal` (internally consistent — + `layout` and `origin_import_id` are derived, not duplicated), `ImportSeal` +- [x] `ImportId` (FNV-1a/64, deterministic) + `reconcile` (marker narrows, + tuple decides; `HashConflict` is permanent) +- [x] `PublicationOutcome` mirroring lance-graph `CommitOutcome` +- [x] `Lifecycle` — **product** of `serving: Option` and + `operation: Operation`, fenced by `Generation`; `LifecycleError` for + stale completions and completions with nothing in flight +- [x] `HeadDecision` — drift AND head-behind (`ReplicaIncomplete`), with + `recover_behind` as the separate recovery decision +- [x] `KnownMaintenance` with `MaintenanceProvenance` (asserted vs verified) +- [x] `StorageRoot` resolution (5 ordered candidates; only `RAILWAY_VOL` and a + probed `/volume01` are intrinsically `Volume`; candidates 1–2 need an + explicit `override_durability` or default to `Ephemeral`) +- [x] `ActivationPolicy` gate +- [x] `plan_publication` (ordered S3 steps, pointer last) +- [x] `VersionInventory` + `GcAuthority` + `RetentionPlan` (`KeepOnly` in + Phases A–C by construction; `Collectable` only in Phase D) +- [x] `warm_verification_reads` (small seals only) +- [x] contract probes for all of the above (see table) +- [x] **correction pass** (2026-08-15, same session) — see § Correction pass; + re-ran the full `osm_lifecycle` test module after + +### Phase B (observational only this session — not started) +- [ ] `OsmArtifactManager` with atomically replaceable `Arc` +- [ ] `/api/osm/status` +- [ ] cgroup memory split + phase deltas +- [ ] listener/`/health` independent of hydration +- [ ] a real `MaintenanceProvenance::LanceHistoryVerified` resolver, OR name it + as still deferred to Phase C at the status-endpoint level + +### Deferred, named +- [ ] braid epic — **the `braid` CLI is not installed in this container**; the + work items above are the importable form. + +## Correction pass (2026-08-15, same session) + +An external review of the first Phase-A draft (relayed via the operator, +explicitly flagged as possibly-stale) found nine structural issues. All nine +were within the three-clock/one-lineage architecture — none required +reopening the architecture itself: + +1. Serving state and import-operation state must be orthogonal — see + § Serving state and import state are orthogonal. +2. Not every explicitly-configured path is a volume — see § Storage policy. +3. The region/bake path-safety boundary needed a validated type, not a + re-checked `String` — see § One lineage per region. +4. Identity inputs (the digest especially) needed to be parsed/canonical, not + stringly-typed — see § Hash choices. +5. `OriginSeal` had two independently-settable copies of `layout` and no + `origin_import_id` — see § Origin seal. +6. Retention was modeled as permission to delete `everything - keep` — see + § Retention. +7. Every "falsifier" was a decision-function test, not a system falsifier — + see the table above. +8. `observed_head < clean` was folded into "drift" instead of being its own + failure mode — see § Head drift, and head-BEHIND. +9. `known_maintenance` needed to carry *provenance*, not just membership — + see `KnownMaintenance`/`MaintenanceProvenance` above; the real resolver is + still deferred, named rather than hidden. + +All nine are implemented in `crates/cockpit-server/src/osm_lifecycle.rs` and +covered by contract probes (module rewritten in full; the module doc's +"What the tests here do and do not prove" section states the probe/falsifier +distinction inline, not only in this plan). The currently-failing test named +in the review request no longer exists in the corrected model — the failure +mode it targeted (`ImportStarted` clobbering `Active`) is now structurally +unreachable because `Operation` cannot touch `serving` at all. + +## Notes + +- `DatasetVersion` is reused from `lance_graph_contract::scheduler`, not + re-declared — the same type lance-graph's contracts use. +- Phase A is **pure**: no I/O, no async, no Lance calls. That is what makes + "pointer last" and "zero traversal on warm boot" testable as decisions + before any wiring exists — and also why every probe in this file proves a + decision, never a running system; see the falsifiers table. diff --git a/crates/cockpit-server/src/main.rs b/crates/cockpit-server/src/main.rs index 10e8881e1..6743bf17e 100644 --- a/crates/cockpit-server/src/main.rs +++ b/crates/cockpit-server/src/main.rs @@ -44,6 +44,7 @@ mod osm_tiles; mod osm; mod osm_features; mod osm_lance; +mod osm_lifecycle; mod osm_slab_hydrate; // ── Embed the Vite build at compile time ───────────────────────────────────── diff --git a/crates/cockpit-server/src/osm_lifecycle.rs b/crates/cockpit-server/src/osm_lifecycle.rs new file mode 100644 index 000000000..9377f1a3e --- /dev/null +++ b/crates/cockpit-server/src/osm_lifecycle.rs @@ -0,0 +1,2437 @@ +//! The OSM/Lance lifecycle — **three independent clocks, one lineage per +//! region**, decided purely. +//! +//! Phase A of `claude-notes/plans/2026-08-15-osm-lance-lifecycle.md`. Every +//! item here is a **pure decision**: no I/O, no async, no Lance calls. That is +//! deliberate and it is what makes properties like *"the S3 pointer is written +//! last"* and *"a warm boot names no data object"* testable as decisions, +//! before any wiring exists to get them wrong. +//! +//! # The correction this module encodes +//! +//! A live Lance dataset is a **versioned lineage, not an immutable +//! directory**. A legitimate append, index build, compaction, reopen or atime +//! change is not corruption, and must not invalidate the imported OSM source. +//! So nothing here hashes a whole dataset directory, and nothing compares its +//! changing object inventory against one static checksum. +//! +//! # The three clocks — never conflated +//! +//! | # | Clock | Carried by | +//! |---|---|---| +//! | 1 | **Semantic import identity** | [`ImportIdentity`], [`OriginSeal`], [`ImportSeal`] | +//! | 2 | **Physical Lance history** | [`ImportSeal::publication_version`], [`ServingVersion`], `observed_head` | +//! | 3 | **Replica progress** | [`StorageRoot`], [`Published::mirrored_through_version`] | +//! +//! Upstream (`lance-graph` #912/#913) states the failure this separation +//! prevents: retrying cycle 1 while the head stood at V5 recorded cycle 1 as +//! *"sealed into V5"*. A later observed head is a **read horizon**, never +//! evidence that the origin changed. +//! +//! # Serving state and import state are ORTHOGONAL +//! +//! [`Lifecycle`] is a product, not a sum: `serving: Option` beside +//! `operation: Operation`. An earlier draft made them one enum, and the +//! consequence was immediate — `ImportStarted` replaced `Active`, so the +//! artifact being served vanished into the import's state and a later failure +//! had nothing to restore. A failed import must leave **both** active and +//! previous-clean untouched, which is only structurally true if the import was +//! never able to overwrite them in the first place. +//! +//! Completions are **fenced by generation**: starting candidate C supersedes +//! in-flight candidate B, and B's late completion is rejected +//! ([`LifecycleError::StaleCompletion`]) instead of activating or failing C. +//! +//! # Hash choices +//! +//! **FNV-1a/64** is the fast physical batch marker ([`ImportId`]) — the same +//! role the upstream Phase-A contract gives it, an idempotency key and not a +//! security property. Because it is only a marker, [`reconcile`] compares the +//! **complete identity tuple** after the marker matches: an FNV collision is a +//! permanent [`ReconcileVerdict::HashConflict`], never a silent reconcile. +//! **SHA-256** stays for externally published SOA manifests and transport +//! verification. No BLAKE3 in this work. +//! +//! # What the tests here do and do not prove +//! +//! Every test below is a **Phase-A contract probe**: it proves a decision +//! function returns the right decision. It does not prove the running system +//! obeys it, because in Phase A there is no running system to obey it — +//! nothing here is wired to `main` yet. Each probe that has a corresponding +//! *system* falsifier names the phase that owes it, and the plan file tracks +//! those as still-open. Confusing the two is how a green suite comes to +//! certify behaviour nobody measured. + +// Phase A (see the module doc) defines the lifecycle VOCABULARY and proves it +// with contract probes; nothing here is wired to `main` yet — that is Phase B +// (`OsmArtifactManager`) and Phase C (versioned import). In a *library* crate +// `pub` alone would exempt these from dead-code analysis; `cockpit-server` is +// a binary, where only items reachable from `main` count as used. Silencing +// the lint is deliberate and scoped to this module, not a blanket workspace +// suppression, and every item it covers already has a probe below exercising +// its logic — the code runs, it is only not yet CALLED by the server. +#![allow(dead_code)] + +use std::collections::BTreeSet; +use std::fmt; +use std::path::PathBuf; + +use lance_graph_contract::scheduler::DatasetVersion; + +// ── Validated identity primitives ──────────────────────────────────────── + +/// Why a value was refused as an identity component. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum IdentityError { + Empty { + field: &'static str, + }, + TooLong { + field: &'static str, + len: usize, + }, + IllegalCharacter { + field: &'static str, + ch: char, + }, + NotHex { + field: &'static str, + }, + WrongLength { + field: &'static str, + want: usize, + got: usize, + }, +} + +impl fmt::Display for IdentityError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Empty { field } => write!(f, "{field} must not be empty"), + Self::TooLong { field, len } => write!(f, "{field} is too long ({len})"), + Self::IllegalCharacter { field, ch } => { + write!(f, "{field} contains an illegal character {ch:?}") + } + Self::NotHex { field } => write!(f, "{field} is not hex"), + Self::WrongLength { field, want, got } => { + write!(f, "{field} must be {want} characters, got {got}") + } + } + } +} + +impl std::error::Error for IdentityError {} + +/// A validated region name. +/// +/// **The validation is not decoration.** This string is interpolated into an +/// S3 object key AND joined onto a filesystem path, so `..`, `/`, or a +/// backslash would let a stray environment variable read a different prefix or +/// write outside the artifact root. The rule is the one the existing slab +/// hydrator already enforces — non-empty, `<= 64` bytes, lowercase +/// `[a-z0-9-]` — restated here as a *type* so the guarantee travels with the +/// value instead of being re-checked (or forgotten) at each use site. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct RegionId(String); + +impl RegionId { + /// The same predicate as `osm_slab_hydrate::is_valid_region`, as a + /// constructor. Rejecting the whole alphabet outside `[a-z0-9-]` makes + /// path traversal and prefix escape impossible by construction rather than + /// by careful escaping at each site. + pub fn parse(s: &str) -> Result { + const FIELD: &str = "region"; + if s.is_empty() { + return Err(IdentityError::Empty { field: FIELD }); + } + if s.len() > 64 { + return Err(IdentityError::TooLong { + field: FIELD, + len: s.len(), + }); + } + if let Some(ch) = s + .chars() + .find(|c| !(c.is_ascii_lowercase() || c.is_ascii_digit() || *c == '-')) + { + return Err(IdentityError::IllegalCharacter { field: FIELD, ch }); + } + Ok(Self(s.to_string())) + } + + #[must_use] + pub fn as_str(&self) -> &str { + &self.0 + } +} + +impl fmt::Display for RegionId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&self.0) + } +} + +/// Which SOA bake a lineage version came from — validated on the same footing +/// as [`RegionId`]. +/// +/// It does not name an S3 key *today*, but it is the kind of value that grows +/// into run ids and staging directory names, and a validated type cannot be +/// quietly promoted into a path. `.` and `_` are allowed (bake ids carry +/// dates and suffixes); `..` is rejected outright. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct SourceBakeId(String); + +impl SourceBakeId { + pub fn parse(s: &str) -> Result { + const FIELD: &str = "source_bake_id"; + if s.is_empty() { + return Err(IdentityError::Empty { field: FIELD }); + } + if s.len() > 128 { + return Err(IdentityError::TooLong { + field: FIELD, + len: s.len(), + }); + } + if let Some(ch) = s.chars().find(|c| { + !(c.is_ascii_lowercase() || c.is_ascii_digit() || *c == '-' || *c == '.' || *c == '_') + }) { + return Err(IdentityError::IllegalCharacter { field: FIELD, ch }); + } + if s.contains("..") { + return Err(IdentityError::IllegalCharacter { + field: FIELD, + ch: '.', + }); + } + Ok(Self(s.to_string())) + } + + #[must_use] + pub fn as_str(&self) -> &str { + &self.0 + } +} + +/// SHA-256 of the externally published SOA manifest for a bake, stored as the +/// **parsed 32 bytes**. +/// +/// Held as bytes, not as whatever string arrived: `A1B2…` and `a1b2…` are the +/// same digest, and if the identity hashed the text instead of the value the +/// same import would derive two different [`ImportId`]s depending on who +/// wrote the manifest. Canonical lowercase hex is a rendering +/// ([`Self::to_hex`]), never the stored form. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct SourceManifestSha256([u8; 32]); + +impl SourceManifestSha256 { + /// Parse 64 hex characters, either case. + pub fn parse_hex(s: &str) -> Result { + const FIELD: &str = "source_manifest_sha256"; + if s.len() != 64 { + return Err(IdentityError::WrongLength { + field: FIELD, + want: 64, + got: s.len(), + }); + } + let mut out = [0u8; 32]; + let raw = s.as_bytes(); + for (i, byte) in out.iter_mut().enumerate() { + let hi = (raw[2 * i] as char) + .to_digit(16) + .ok_or(IdentityError::NotHex { field: FIELD })?; + let lo = (raw[2 * i + 1] as char) + .to_digit(16) + .ok_or(IdentityError::NotHex { field: FIELD })?; + *byte = ((hi << 4) | lo) as u8; + } + Ok(Self(out)) + } + + #[must_use] + pub fn from_bytes(bytes: [u8; 32]) -> Self { + Self(bytes) + } + + #[must_use] + pub fn as_bytes(&self) -> &[u8; 32] { + &self.0 + } + + /// Canonical lowercase hex — the published form. + #[must_use] + pub fn to_hex(self) -> String { + let mut s = String::with_capacity(64); + for b in &self.0 { + s.push_str(&format!("{b:02x}")); + } + s + } +} + +// ── Clock 1: semantic import identity ──────────────────────────────────── + +/// The row layout a lineage was written with. +/// +/// Only an **incompatible** layout mints a new lineage. A new SOA bake under +/// the same ABI is a new sealed import *inside* the existing lineage — not a +/// new dataset prefix. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct LayoutAbi(pub u32); + +/// The writer implementation's binary ABI, recorded so a reader can refuse a +/// lineage written by an incompatible producer even when the layout matches. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct WriterAbi(pub u32); + +/// The current layout: `osm-lance-v2`. +pub const LAYOUT_V2: LayoutAbi = LayoutAbi(2); + +/// A stable Lance lineage: one region at one layout ABI. +/// +/// Renders as `berlin/osm-lance-v2` — the S3 path segment and the identity in +/// one value, so the two cannot drift apart. The region is a [`RegionId`], so +/// the rendered segment is path-safe by construction. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct LineageId { + pub region: RegionId, + pub layout: LayoutAbi, +} + +impl LineageId { + #[must_use] + pub fn new(region: RegionId, layout: LayoutAbi) -> Self { + Self { region, layout } + } + + /// Convenience: validate a region string and build the lineage in one step. + pub fn parse(region: &str, layout: LayoutAbi) -> Result { + Ok(Self::new(RegionId::parse(region)?, layout)) + } + + /// `/osm-lance-v` — the lineage's S3 path segment. + #[must_use] + pub fn path_segment(&self) -> String { + format!("{}/osm-lance-v{}", self.region, self.layout.0) + } + + /// Whether a dataset written at `theirs` can be served by a reader that + /// expects this lineage's layout. + /// + /// Exact equality, deliberately: "compatible" for a row layout means the + /// same carving, and a reader that guesses at near-misses is how a silent + /// misread ships. + #[must_use] + pub fn accepts_layout(&self, theirs: LayoutAbi) -> bool { + self.layout == theirs + } +} + +/// The deterministic physical batch marker — FNV-1a/64 over the identity that +/// defines *this* import. +/// +/// It is an **idempotency key**, not a digest of the data and not a +/// cryptographic commitment: two runs of the same bake into the same lineage +/// at the same ABIs must produce the same `ImportId`, so a retry after a lost +/// acknowledgement reconciles instead of publishing a second version. Because +/// it is 64 bits and non-cryptographic, a match is a *candidate* — see +/// [`reconcile`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct ImportId(pub u64); + +impl ImportId { + /// Hex, zero-padded — the form used in the `import-seals/.yaml` key. + /// Hex is key-safe by construction; no validation needed at the use site. + #[must_use] + pub fn to_key(self) -> String { + format!("{:016x}", self.0) + } +} + +const FNV_OFFSET: u64 = 0xcbf2_9ce4_8422_2325; +const FNV_PRIME: u64 = 0x0000_0100_0000_01b3; + +fn fnv1a64(bytes: &[u8], seed: u64) -> u64 { + let mut h = seed; + for b in bytes { + h ^= u64::from(*b); + h = h.wrapping_mul(FNV_PRIME); + } + h +} + +/// The complete semantic identity of one import — the tuple, not the marker. +/// +/// `(lineage[region + layout_abi], bake, source_manifest_sha256, writer_abi)`. +/// [`ImportId`] is derived from it; reconciliation compares **this**, because +/// a 64-bit marker match is not proof of sameness. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ImportIdentity { + pub lineage: LineageId, + pub bake: SourceBakeId, + pub manifest: SourceManifestSha256, + pub writer: WriterAbi, +} + +impl ImportIdentity { + #[must_use] + pub fn new( + lineage: LineageId, + bake: SourceBakeId, + manifest: SourceManifestSha256, + writer: WriterAbi, + ) -> Self { + Self { + lineage, + bake, + manifest, + writer, + } + } + + #[must_use] + pub fn layout(&self) -> LayoutAbi { + self.lineage.layout + } + + /// Derive the physical batch marker. + /// + /// A separator byte is folded between fields so that `("ab", "c")` and + /// `("a", "bc")` cannot collide — concatenating variable-length fields + /// without one is the classic way to build a hash that looks deterministic + /// and is quietly ambiguous. The manifest contributes its **32 parsed + /// bytes**, so hex casing cannot change the marker. + #[must_use] + pub fn import_id(&self) -> ImportId { + let mut h = FNV_OFFSET; + for field in [ + self.lineage.region.as_str().as_bytes(), + &self.lineage.layout.0.to_le_bytes(), + self.bake.as_str().as_bytes(), + self.manifest.as_bytes().as_slice(), + &self.writer.0.to_le_bytes(), + ] { + h = fnv1a64(field, h); + h = fnv1a64(b"\x1f", h); // unit separator — never part of a field + } + ImportId(h) + } +} + +/// What a marker match means once the full tuple is compared. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ReconcileVerdict { + /// Same marker AND same tuple: genuinely the same import. Reconcile — do + /// not create a second version. + SameImport, + /// Same marker, **different tuple**: an FNV-1a/64 collision. This is a + /// permanent identity conflict (upstream's `CommitError::HashConflict`), + /// never a reconcile — reconciling here would adopt someone else's data + /// as this import's. + HashConflict { + marker: ImportId, + candidate: Box, + found: Box, + }, + /// Different marker: not a retry of this import at all. + DifferentImport, +} + +/// Compare a candidate import against one found in history under the same +/// marker. +/// +/// The marker narrows the search; the **tuple** decides. That ordering is the +/// whole reason a 64-bit non-cryptographic marker is safe to use here. +#[must_use] +pub fn reconcile(candidate: &ImportIdentity, found: &ImportIdentity) -> ReconcileVerdict { + let marker = candidate.import_id(); + if marker != found.import_id() { + return ReconcileVerdict::DifferentImport; + } + if candidate == found { + ReconcileVerdict::SameImport + } else { + ReconcileVerdict::HashConflict { + marker, + candidate: Box::new(candidate.clone()), + found: Box::new(found.clone()), + } + } +} + +/// The small, immutable identity of a lineage's origin. +/// +/// Mirrored to S3 beside the lineage. A warm boot verifies **this** — a few +/// hundred bytes — plus compatible Lance metadata. It must never require +/// checking out and reading all V1 data to prove the origin. +/// +/// Fields are private and the only constructor takes an [`ImportIdentity`], so +/// a seal cannot contradict itself: the layout is the lineage's layout (there +/// is no second unconstrained copy to drift), and `origin_import_id` is +/// *derived*, so the seal always identifies an import that could actually have +/// produced it. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct OriginSeal { + identity: ImportIdentity, + origin_import_id: ImportId, + origin_publication: Option, +} + +impl OriginSeal { + /// Seal a lineage's origin from the import that created it. + #[must_use] + pub fn create(identity: ImportIdentity, origin_publication: Option) -> Self { + let origin_import_id = identity.import_id(); + Self { + identity, + origin_import_id, + origin_publication, + } + } + + #[must_use] + pub fn lineage(&self) -> &LineageId { + &self.identity.lineage + } + /// The layout — read from the lineage, never stored twice. + #[must_use] + pub fn layout(&self) -> LayoutAbi { + self.identity.lineage.layout + } + #[must_use] + pub fn writer(&self) -> WriterAbi { + self.identity.writer + } + #[must_use] + pub fn initial_bake(&self) -> &SourceBakeId { + &self.identity.bake + } + #[must_use] + pub fn initial_manifest(&self) -> &SourceManifestSha256 { + &self.identity.manifest + } + /// The import that created the lineage — without this the first sealed + /// import cannot be identified at all. + #[must_use] + pub fn origin_import_id(&self) -> ImportId { + self.origin_import_id + } + /// Where the origin import landed. `Some` only if that publication was + /// observed fresh; see [`ImportSeal::publication_version`]. + #[must_use] + pub fn origin_publication(&self) -> Option { + self.origin_publication + } + + /// Whether a later import belongs to this lineage at a compatible ABI. + #[must_use] + pub fn accepts(&self, other: &ImportIdentity) -> bool { + self.identity.lineage == other.lineage && self.identity.writer == other.writer + } +} + +/// The result of one semantically successful import. +/// +/// Physical Lance maintenance versions may advance the head **without** +/// creating a new `ImportSeal` — that is the whole point of separating clock 1 +/// from clock 2. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ImportSeal { + pub identity: ImportIdentity, + pub import_id: ImportId, + /// The PHYSICAL publication position — `Some` **only** when this call + /// published fresh. + /// + /// `None` on a reconciled retry, mirroring `lance-graph`'s `SealedCycle`: + /// a reconciled batch was already durable, its original position is not + /// known from the outcome, and it is never invented here. The durable + /// identity is the [`ImportId`]; the position is recovered by searching + /// that id in Lance history on the audit path. + pub publication_version: Option, + /// The head actually OBSERVED by this outcome — a read horizon in every + /// case, never an artifact's publication position. + pub observed_head: Option, +} + +/// What a publication attempt actually did — the q2-side mirror of +/// `lance_graph_planner::persist_sink::CommitOutcome`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum PublicationOutcome { + /// Published fresh at this version. + Committed(DatasetVersion), + /// Already durable under the same [`ImportId`] **and the same identity + /// tuple** ([`reconcile`] ran first); no second version was created. + /// `current_head` is the head **at reconciliation time**. + Reconciled { current_head: DatasetVersion }, + /// The outcome could not be determined. Re-submit the SAME frozen import + /// identity — reconciliation runs first, so the retry cannot double-write. + /// There is **no compensating delete**. + Ambiguous, +} + +impl ImportSeal { + /// Seal an outcome, applying the publication/observation rule. + #[must_use] + pub fn seal(identity: ImportIdentity, outcome: PublicationOutcome) -> Option { + let (publication_version, observed_head) = match outcome { + PublicationOutcome::Committed(v) => (Some(v), Some(v)), + PublicationOutcome::Reconciled { current_head } => (None, Some(current_head)), + // Ambiguous seals nothing: the identity is re-submitted frozen. + PublicationOutcome::Ambiguous => return None, + }; + let import_id = identity.import_id(); + Some(Self { + identity, + import_id, + publication_version, + observed_head, + }) + } +} + +// ── Clock 2: physical Lance history ────────────────────────────────────── + +/// What this replica serves, and from which version. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ServingVersion { + pub version: DatasetVersion, + /// `true` when the served version is a checkout of a known clean sealed + /// version rather than the current head. + pub pinned_to_clean: bool, +} + +/// Where a claim that "version V is ordinary maintenance" comes from. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MaintenanceProvenance { + /// **Phase-A limitation, named rather than hidden.** The caller asserted + /// this set; nothing has verified it against Lance transaction history. A + /// caller that asserts a hostile version would have it served. The real + /// resolver — walk the transaction log, confirm each version is a + /// compaction/index/deletion-materialisation descendant of `clean` — is + /// deferred to Phase C, and until it exists the status endpoint must + /// report the decision as *asserted*, not *verified*. + CallerAsserted, + /// Resolved by walking Lance history and confirming each version is a + /// maintenance descendant of the clean publication. + LanceHistoryVerified, +} + +/// Versions recognised as maintenance descendants, **with provenance**. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct KnownMaintenance { + versions: BTreeSet, + provenance: MaintenanceProvenance, +} + +impl KnownMaintenance { + /// Phase A / Phase B: whatever the caller says, carried as asserted. + #[must_use] + pub fn asserted(versions: BTreeSet) -> Self { + Self { + versions, + provenance: MaintenanceProvenance::CallerAsserted, + } + } + + /// Phase C+: produced by a real history walk. + #[must_use] + pub fn verified(versions: BTreeSet) -> Self { + Self { + versions, + provenance: MaintenanceProvenance::LanceHistoryVerified, + } + } + + #[must_use] + pub fn none() -> Self { + Self::asserted(BTreeSet::new()) + } + + #[must_use] + pub fn contains(&self, v: DatasetVersion) -> bool { + self.versions.contains(&v) + } + + #[must_use] + pub fn provenance(&self) -> MaintenanceProvenance { + self.provenance + } +} + +/// Why the replica cannot serve the clean version it wanted. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum BehindReason { + /// The observed head is *earlier* than the last clean import: this replica + /// is truncated or stale. Not drift — the head did not advance, the + /// replica fell behind. + HeadBehindClean, + /// The head is fine, but the clean version itself is not present locally, + /// so it cannot be checked out. + CleanVersionAbsent, +} + +/// How to serve, given the last clean publication and what this replica has. +/// +/// Never "discard the lineage": a head that advanced is ordinary Lance life, +/// and a head that is *behind* is a replica problem, not an origin problem. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum HeadDecision { + /// Head equals the clean publication — serve it directly. + ServeHead(DatasetVersion), + /// The head is a recognised maintenance descendant of the clean version + /// (compaction, index build): serve it, no drift reported. The provenance + /// travels with the decision so the status endpoint can say whether that + /// recognition was *verified* or merely *asserted*. + ServeKnownDescendant { + version: DatasetVersion, + provenance: MaintenanceProvenance, + }, + /// The head advanced for an unknown reason **and the clean version is + /// locally available**: serve a checkout of the last known clean sealed + /// version and report drift. + CheckoutClean { + clean: DatasetVersion, + head: DatasetVersion, + }, + /// The replica cannot produce the clean version at all. This is a + /// *recovery* case, not a serving case — see [`recover_behind`]. Issuing a + /// checkout here would request a version the replica may not possess. + ReplicaIncomplete { + clean: DatasetVersion, + head: DatasetVersion, + reason: BehindReason, + }, +} + +impl HeadDecision { + #[must_use] + pub fn head_drift(self) -> bool { + matches!(self, Self::CheckoutClean { .. }) + } + + /// The version to serve, when there is one. `None` for + /// [`Self::ReplicaIncomplete`] — that state has nothing to serve until + /// recovery runs. + #[must_use] + pub fn serving(self) -> Option { + match self { + Self::ServeHead(v) | Self::ServeKnownDescendant { version: v, .. } => { + Some(ServingVersion { + version: v, + pinned_to_clean: false, + }) + } + Self::CheckoutClean { clean, .. } => Some(ServingVersion { + version: clean, + pinned_to_clean: true, + }), + Self::ReplicaIncomplete { .. } => None, + } + } +} + +/// What this replica knows about a lineage right now. +#[derive(Debug, Clone)] +pub struct HeadFacts<'a> { + /// The last known clean sealed publication. + pub clean: DatasetVersion, + /// The head this replica observes. + pub observed_head: DatasetVersion, + pub known_maintenance: &'a KnownMaintenance, + /// Versions this replica can actually check out. Checking out anything + /// outside this set is a request the replica may be unable to satisfy. + pub locally_available: &'a BTreeSet, +} + +/// Decide what to serve. +#[must_use] +pub fn decide_head(facts: &HeadFacts<'_>) -> HeadDecision { + let HeadFacts { + clean, + observed_head, + known_maintenance, + locally_available, + } = *facts; + + if observed_head < clean { + return HeadDecision::ReplicaIncomplete { + clean, + head: observed_head, + reason: BehindReason::HeadBehindClean, + }; + } + if observed_head == clean { + return if locally_available.contains(&clean) { + HeadDecision::ServeHead(clean) + } else { + HeadDecision::ReplicaIncomplete { + clean, + head: observed_head, + reason: BehindReason::CleanVersionAbsent, + } + }; + } + if known_maintenance.contains(observed_head) { + return HeadDecision::ServeKnownDescendant { + version: observed_head, + provenance: known_maintenance.provenance(), + }; + } + if locally_available.contains(&clean) { + HeadDecision::CheckoutClean { + clean, + head: observed_head, + } + } else { + HeadDecision::ReplicaIncomplete { + clean, + head: observed_head, + reason: BehindReason::CleanVersionAbsent, + } + } +} + +/// What to do when the replica cannot serve the clean version. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum BehindRecovery { + /// S3 holds the clean version: hydrate it. The preferred path — it + /// restores the intended artifact rather than silently serving an older + /// one. + HydrateFromS3 { want: DatasetVersion }, + /// S3 cannot supply it, but this replica does hold an older clean version: + /// serve that, degraded but honest. + ServeAvailablePrevious { version: DatasetVersion }, + /// Nothing serviceable. The status endpoint reports it; nothing is + /// deleted, and the lineage is not discarded. + Unserviceable, +} + +/// Recover from [`HeadDecision::ReplicaIncomplete`]. +/// +/// `available_clean` is the set of *sealed clean* versions this replica can +/// actually check out — not every version it has, because serving an +/// unverified version to escape a recovery is how a replica ends up quietly +/// serving something no seal describes. +#[must_use] +pub fn recover_behind( + clean: DatasetVersion, + available_clean: &BTreeSet, + s3_has_clean: bool, +) -> BehindRecovery { + if s3_has_clean { + return BehindRecovery::HydrateFromS3 { want: clean }; + } + match available_clean.iter().rev().find(|v| **v < clean) { + Some(v) => BehindRecovery::ServeAvailablePrevious { version: *v }, + None => BehindRecovery::Unserviceable, + } +} + +// ── Clock 3: replica progress ──────────────────────────────────────────── + +/// Whether the resolved artifact root survives a redeploy. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RootKind { + /// A mounted volume — artifacts survive a rebuild. + Volume, + /// Container-local disk — everything is re-fetched on the next deploy. + Ephemeral, +} + +/// The resolved artifact root and its durability. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct StorageRoot { + pub path: PathBuf, + pub kind: RootKind, + /// Which policy candidate supplied it, for the status endpoint. + pub source: &'static str, +} + +/// The ordered storage policy. Pure: the caller supplies what it observed of +/// the environment, so the ordering is testable without touching a filesystem. +/// +/// 1. `OSM_ARTIFACT_ROOT` +/// 2. `OSM_SLAB_CACHE_DIR` (the existing variable — kept, not renamed) +/// 3. `RAILWAY_VOL/osm` +/// 4. `/volume01/osm` when present and writable +/// 5. documented ephemeral fallback +#[derive(Debug, Clone)] +pub struct RootInputs { + pub osm_artifact_root: Option, + pub osm_slab_cache_dir: Option, + /// What the operator declares about candidates 1 and 2. + /// + /// **A generic override says nothing about durability.** `OSM_ARTIFACT_ROOT` + /// and `OSM_SLAB_CACHE_DIR` routinely point at container-local scratch — + /// treating "an operator named it" as "an operator guaranteed it survives" + /// is exactly the assumption that lets an ephemeral candidate activate + /// without S3 and re-convert the whole bake on the next deploy. Absent an + /// explicit declaration this is `None`, and `None` means + /// [`RootKind::Ephemeral`] — the fail-safe direction, because being + /// wrongly cautious costs one S3 wait and being wrongly confident costs a + /// full re-import. + pub override_durability: Option, + pub railway_vol: Option, + /// Whether `/volume01` exists AND is writable — the caller probes it. + pub volume01_writable: bool, + /// Where the ephemeral fallback lands. Never empty and never relative — + /// see [`RootInputs::default`]. + pub ephemeral_dir: PathBuf, +} + +/// The default ephemeral directory leaf, under the platform temp dir. +const EPHEMERAL_LEAF: &str = "q2-osm"; + +impl Default for RootInputs { + /// The fallback is an **absolute** platform temp path. + /// + /// A `PathBuf::default()` is empty, and an empty path resolves against the + /// process's current directory — so a defaulted `RootInputs` would write + /// the artifact tree into whatever directory the server happened to be + /// started from. `std::env::temp_dir()` is the cross-platform absolute + /// answer (`/tmp` on Linux, `%TEMP%` on Windows) and never returns an + /// empty path. + fn default() -> Self { + Self { + osm_artifact_root: None, + osm_slab_cache_dir: None, + override_durability: None, + railway_vol: None, + volume01_writable: false, + ephemeral_dir: std::env::temp_dir().join(EPHEMERAL_LEAF), + } + } +} + +/// A configured path is usable only if it is non-blank AND absolute. +/// +/// A relative override resolves against the current directory, which is the +/// same hazard as an empty one — the artifact tree lands wherever the process +/// was launched. Both are treated as *absent*, so the ordered policy falls +/// through to the next candidate instead of silently accepting a path that +/// means different things to different processes. +fn usable_path(s: &Option) -> Option { + let trimmed = s.as_ref()?.trim(); + if trimmed.is_empty() { + return None; + } + let p = PathBuf::from(trimmed); + p.is_absolute().then_some(p) +} + +/// Resolve the artifact root. +/// +/// Candidates 1 and 2 take their durability from +/// [`RootInputs::override_durability`], defaulting to +/// [`RootKind::Ephemeral`]. Only `RAILWAY_VOL` and a probed-writable +/// `/volume01` are *intrinsically* known to be volume storage. +#[must_use] +pub fn resolve_root(inputs: &RootInputs) -> StorageRoot { + let override_kind = inputs.override_durability.unwrap_or(RootKind::Ephemeral); + + if let Some(p) = usable_path(&inputs.osm_artifact_root) { + return StorageRoot { + path: p, + kind: override_kind, + source: "OSM_ARTIFACT_ROOT", + }; + } + if let Some(p) = usable_path(&inputs.osm_slab_cache_dir) { + return StorageRoot { + path: p, + kind: override_kind, + source: "OSM_SLAB_CACHE_DIR", + }; + } + if let Some(p) = usable_path(&inputs.railway_vol) { + return StorageRoot { + path: p.join("osm"), + kind: RootKind::Volume, + source: "RAILWAY_VOL/osm", + }; + } + if inputs.volume01_writable { + return StorageRoot { + path: PathBuf::from("/volume01/osm"), + kind: RootKind::Volume, + source: "/volume01/osm", + }; + } + // Last line of defence: even a hand-built `RootInputs` must not resolve to + // "" or a relative path. + let ephemeral = + if inputs.ephemeral_dir.as_os_str().is_empty() || !inputs.ephemeral_dir.is_absolute() { + std::env::temp_dir().join(EPHEMERAL_LEAF) + } else { + inputs.ephemeral_dir.clone() + }; + StorageRoot { + path: ephemeral, + kind: RootKind::Ephemeral, + source: "ephemeral fallback", + } +} + +/// The S3 publication pointer for a lineage — written **last**. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Published { + pub active_clean: ImportId, + pub previous_clean: Option, + /// How far this replica has mirrored the lineage to S3. + pub mirrored_through_version: Option, +} + +// ── Activation ─────────────────────────────────────────────────────────── + +/// Whether a locally verified candidate may serve before S3 has it. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ActivationPolicy { + pub volume_requires_s3: bool, + pub ephemeral_requires_s3: bool, +} + +impl Default for ActivationPolicy { + /// The documented default: a volume may run ahead of S3, an ephemeral root + /// may not. + fn default() -> Self { + Self { + volume_requires_s3: false, + ephemeral_requires_s3: true, + } + } +} + +/// The verdict on activating a locally verified candidate. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Activation { + /// Serve it; S3 already has it. + Activate, + /// Serve it, and report that S3 is behind. Legitimate on a volume: the + /// artifact survives a redeploy, so a lagging mirror costs nothing but + /// rollback distance. + ActivateS3Lagging, + /// Do NOT serve yet. On an ephemeral root an un-mirrored candidate is not + /// fully successful: the next redeploy would repeat the whole SOA→Lance + /// conversion, which is the cost this design exists to remove. + HoldForS3, +} + +#[must_use] +pub fn decide_activation( + kind: RootKind, + s3_has_candidate: bool, + policy: ActivationPolicy, +) -> Activation { + if s3_has_candidate { + return Activation::Activate; + } + let requires = match kind { + RootKind::Volume => policy.volume_requires_s3, + RootKind::Ephemeral => policy.ephemeral_requires_s3, + }; + if requires { + Activation::HoldForS3 + } else { + Activation::ActivateS3Lagging + } +} + +// ── Lifecycle: serving state × import operation ────────────────────────── + +/// Monotonic import attempt counter — the fence. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct Generation(pub u64); + +/// What this replica currently serves. Independent of any import in flight. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ServingSet { + pub active: ImportId, + /// Retained for rollback. Only ONE previous — never a growing chain. + pub previous: Option, +} + +/// Why a candidate import failed, classified for the status endpoint. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ImportFailure { + /// The SOA source could not be read or was rejected. + Source(String), + /// The build/commit failed. + Write(String), + /// The candidate committed but failed validation. + Validation(String), + /// The candidate is fine locally but could not be mirrored, and this root + /// requires S3 before activation. + S3Publication(String), + /// A permanent identity conflict — see [`ReconcileVerdict::HashConflict`]. + IdentityConflict, +} + +/// The import operation, orthogonal to what is being served. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Operation { + Idle, + Importing { + candidate: ImportId, + generation: Generation, + }, + Failed { + candidate: ImportId, + generation: Generation, + error: ImportFailure, + }, +} + +/// A refused transition. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum LifecycleError { + /// A completion arrived for an attempt that is no longer in flight — + /// typically B's late success after C superseded it. Applying it would + /// activate or fail the wrong candidate. + StaleCompletion { + candidate: ImportId, + generation: Generation, + current: Option, + }, + /// A completion arrived with no import in flight at all. + NoImportInFlight { candidate: ImportId }, +} + +impl fmt::Display for LifecycleError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::StaleCompletion { + candidate, + generation, + current, + } => write!( + f, + "stale completion for {} at generation {} (current {:?})", + candidate.to_key(), + generation.0, + current.map(|g| g.0) + ), + Self::NoImportInFlight { candidate } => { + write!(f, "no import in flight for {}", candidate.to_key()) + } + } + } +} + +impl std::error::Error for LifecycleError {} + +/// The lifecycle: a **product** of serving state and import state. +/// +/// Nothing an import does can reach `serving` except a successful activation. +/// That is a structural guarantee, not a rule someone has to remember: the +/// failure paths do not take `serving` as an argument. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Lifecycle { + pub serving: Option, + pub operation: Operation, + next_generation: u64, +} + +impl Lifecycle { + /// Nothing local, nothing in flight. + #[must_use] + pub fn absent() -> Self { + Self { + serving: None, + operation: Operation::Idle, + next_generation: 1, + } + } + + /// A replica that already serves something (e.g. recovered from S3 at boot). + #[must_use] + pub fn serving(active: ImportId, previous: Option) -> Self { + Self { + serving: Some(ServingSet { active, previous }), + operation: Operation::Idle, + next_generation: 1, + } + } + + #[must_use] + pub fn active(&self) -> Option { + self.serving.map(|s| s.active) + } + + #[must_use] + pub fn previous(&self) -> Option { + self.serving.and_then(|s| s.previous) + } + + /// The generation of the import currently in flight, if any. + #[must_use] + pub fn in_flight(&self) -> Option { + match self.operation { + Operation::Importing { generation, .. } => Some(generation), + _ => None, + } + } + + /// Begin a candidate import. Returns the fence token the completion must + /// present. + /// + /// Starting C while B is in flight **supersedes** B: the generation + /// advances, and B's later completion is refused. Serving is untouched. + #[must_use] + pub fn start_import(&self, candidate: ImportId) -> (Self, Generation) { + let generation = Generation(self.next_generation); + ( + Self { + serving: self.serving, + operation: Operation::Importing { + candidate, + generation, + }, + next_generation: self.next_generation + 1, + }, + generation, + ) + } + + /// The candidate is verified locally AND cleared its activation gate. + /// + /// Rotation happens here and only here: the outgoing active becomes + /// `previous`, and the chain never grows beyond one. + pub fn activate( + &self, + candidate: ImportId, + generation: Generation, + ) -> Result { + self.check_fence(candidate, generation)?; + let previous = self.active(); + Ok(Self { + serving: Some(ServingSet { + active: candidate, + previous, + }), + operation: Operation::Idle, + next_generation: self.next_generation, + }) + } + + /// The candidate failed. + /// + /// `serving` is copied through verbatim — a failed import cannot touch + /// either the active or the previous-clean artifact. + pub fn fail( + &self, + candidate: ImportId, + generation: Generation, + error: ImportFailure, + ) -> Result { + self.check_fence(candidate, generation)?; + Ok(Self { + serving: self.serving, + operation: Operation::Failed { + candidate, + generation, + error, + }, + next_generation: self.next_generation, + }) + } + + fn check_fence( + &self, + candidate: ImportId, + generation: Generation, + ) -> Result<(), LifecycleError> { + match self.operation { + Operation::Importing { + candidate: c, + generation: g, + } if c == candidate && g == generation => Ok(()), + Operation::Importing { generation: g, .. } => Err(LifecycleError::StaleCompletion { + candidate, + generation, + current: Some(g), + }), + _ => Err(LifecycleError::NoImportInFlight { candidate }), + } + } +} + +// ── Ordered plans ──────────────────────────────────────────────────────── + +/// One step of the S3 publication sequence. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum PublishStep { + /// Ensure the outgoing active is durable as `previous_clean` first. + EnsurePreviousDurable(ImportId), + /// Upload only the objects S3 does not already hold. + UploadMissingObjects(Vec), + /// The small seals. + UploadSeal(String), + /// The head pointer — **always last**, with generation/ETag protection. + PublishPointer { + active: ImportId, + previous: Option, + }, +} + +/// Build the ordered publication plan. +/// +/// `already_in_s3` is the set of object keys the lineage already holds: +/// natively referenced Lance fragments are **never uploaded twice**. +#[must_use] +pub fn plan_publication( + candidate: ImportId, + outgoing_active: Option, + candidate_objects: &[String], + already_in_s3: &BTreeSet, + seal_keys: &[String], +) -> Vec { + let mut steps = Vec::new(); + if let Some(prev) = outgoing_active { + steps.push(PublishStep::EnsurePreviousDurable(prev)); + } + let missing: Vec = candidate_objects + .iter() + .filter(|k| !already_in_s3.contains(*k)) + .cloned() + .collect(); + steps.push(PublishStep::UploadMissingObjects(missing)); + for s in seal_keys { + steps.push(PublishStep::UploadSeal(s.clone())); + } + steps.push(PublishStep::PublishPointer { + active: candidate, + previous: outgoing_active, + }); + steps +} + +/// What a **warm boot** is allowed to read to prove the origin. +/// +/// Small seals and Lance metadata only. The absence of any data object here is +/// the point: proving an origin must never require checking out and reading +/// the imported map data. +#[must_use] +pub fn warm_verification_reads(lineage: &LineageId, active: ImportId) -> Vec { + let base = format!("q2/lance/{}", lineage.path_segment()); + vec![ + format!("{base}/origin-seal.yaml"), + format!("{base}/published.yaml"), + format!("{base}/import-seals/{}.yaml", active.to_key()), + ] +} + +// ── Retention ──────────────────────────────────────────────────────────── + +/// Everything reachable from one pinned Lance version. +/// +/// "Reachable" means the **whole closure**: the manifest, every data fragment, +/// index files, deletion files, transaction records and any metadata the +/// version needs to open. An object list that is merely "the fragments I +/// uploaded" is not an inventory, and treating it as one is how retention +/// deletes a deletion-file and turns a clean version into an unopenable one. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct VersionInventory { + pub version: DatasetVersion, + pub objects: BTreeSet, + /// `false` when the caller could not enumerate the closure completely. + /// An incomplete inventory can never authorise a deletion. + pub complete: bool, +} + +impl VersionInventory { + #[must_use] + pub fn complete(version: DatasetVersion, objects: BTreeSet) -> Self { + Self { + version, + objects, + complete: true, + } + } + + #[must_use] + pub fn partial(version: DatasetVersion, objects: BTreeSet) -> Self { + Self { + version, + objects, + complete: false, + } + } +} + +/// Who vouches that an inventory is a real Lance reachability closure. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum GcAuthority { + /// Phases A–C: **nothing may be deleted.** No mechanism exists yet that + /// can prove an object is unreachable from every retained version, and + /// `everything - keep` is not that proof — it is a guess dressed as a set + /// operation. Retention in these phases is *intent*, and intent does not + /// delete. + None, + /// Phase D: a Lance-native reachability/cleanup mechanism produced these + /// inventories. + LanceNativeReachability, +} + +/// Why a retention plan authorises no deletion. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RetentionHold { + /// The grace period has not elapsed. + GracePeriodActive, + /// At least one inventory is incomplete. + IncompleteInventory, + /// No Lance-native reachability mechanism has vouched for the inventories. + NoGcAuthority, +} + +/// The retention decision for one lineage. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RetentionPlan { + /// Intent only. `keep` is what must survive; **no deletion is + /// authorised.** + KeepOnly { + keep: BTreeSet, + hold: RetentionHold, + }, + /// Phase D and later: deletion is authorised for `collect`. + Collectable { + keep: BTreeSet, + collect: BTreeSet, + }, +} + +impl RetentionPlan { + /// The objects this plan authorises deleting — empty unless the plan is + /// [`Self::Collectable`]. + #[must_use] + pub fn collect(&self) -> BTreeSet { + match self { + Self::KeepOnly { .. } => BTreeSet::new(), + Self::Collectable { collect, .. } => collect.clone(), + } + } + + #[must_use] + pub fn keep(&self) -> &BTreeSet { + match self { + Self::KeepOnly { keep, .. } | Self::Collectable { keep, .. } => keep, + } + } +} + +/// Decide retention. +/// +/// Keeps everything reachable from the active-clean and previous-clean +/// versions, plus **all** small seals — seals are tiny and are the only record +/// of what a version meant. +/// +/// It does **not** conclude "delete `everything - keep`". Three independent +/// gates must all open before a single object is proposed for deletion: the +/// grace period must have elapsed, both inventories must be complete, and a +/// [`GcAuthority`] must vouch that they are real reachability closures. In +/// Phase A the last gate is always shut. +#[must_use] +pub fn plan_retention( + active: &VersionInventory, + previous: Option<&VersionInventory>, + all_seals: &BTreeSet, + everything: &BTreeSet, + grace_elapsed: bool, + authority: GcAuthority, +) -> RetentionPlan { + let mut keep: BTreeSet = active.objects.clone(); + if let Some(p) = previous { + keep.extend(p.objects.iter().cloned()); + } + keep.extend(all_seals.iter().cloned()); + + let inventories_complete = active.complete && previous.is_none_or(|p| p.complete); + + let hold = if !inventories_complete { + Some(RetentionHold::IncompleteInventory) + } else if !grace_elapsed { + Some(RetentionHold::GracePeriodActive) + } else if authority == GcAuthority::None { + Some(RetentionHold::NoGcAuthority) + } else { + None + }; + + match hold { + Some(hold) => RetentionPlan::KeepOnly { keep, hold }, + None => { + let collect = everything.difference(&keep).cloned().collect(); + RetentionPlan::Collectable { keep, collect } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn lineage() -> LineageId { + LineageId::parse("berlin", LAYOUT_V2).expect("berlin is a valid region") + } + fn bake() -> SourceBakeId { + SourceBakeId::parse("berlin-260805").expect("valid bake id") + } + fn manifest() -> SourceManifestSha256 { + SourceManifestSha256::parse_hex(&"a".repeat(64)).expect("valid digest") + } + fn identity() -> ImportIdentity { + ImportIdentity::new(lineage(), bake(), manifest(), WriterAbi(1)) + } + fn v(n: u64) -> DatasetVersion { + DatasetVersion(n) + } + fn versions(list: &[u64]) -> BTreeSet { + list.iter().copied().map(DatasetVersion).collect() + } + + // ── Validated identity ──────────────────────────────────────────────── + + /// PROBE (Phase A). The region is interpolated into an S3 key and joined + /// onto a path; these are the values that must never reach either. + #[test] + fn region_rejects_everything_that_could_escape_a_prefix_or_a_directory() { + for bad in [ + "", "..", "../etc", "a/b", "a\\b", "Berlin", "berlin ", "ber_lin", "berlin/", + ] { + assert!( + RegionId::parse(bad).is_err(), + "{bad:?} must not be accepted as a region" + ); + } + assert!( + RegionId::parse(&"a".repeat(65)).is_err(), + "length is bounded" + ); + + // Can-it-accept half: the real names must still work, or the guard is + // just a refusal machine. + for good in ["berlin", "brandenburg", "baden-wuerttemberg", "nrw2"] { + assert!(RegionId::parse(good).is_ok(), "{good:?} must be accepted"); + } + } + + /// PROBE (Phase A). Bake ids grow into staging directory names; `..` in + /// one would be a traversal the moment that happens. + #[test] + fn bake_id_rejects_traversal_and_separators() { + for bad in ["", "..", "a/../b", "a/b", "a\\b", "Berlin-2608"] { + assert!( + SourceBakeId::parse(bad).is_err(), + "{bad:?} must not be accepted as a bake id" + ); + } + for good in ["berlin-260805", "berlin.v2_final", "bb-260901"] { + assert!( + SourceBakeId::parse(good).is_ok(), + "{good:?} must be accepted" + ); + } + } + + /// PROBE (Phase A). The same digest in different casing is the same + /// digest — and therefore the same import. Hashing the *text* would derive + /// two identities for one bake and publish it twice. + #[test] + fn digest_casing_cannot_change_the_import_identity() { + let lower = SourceManifestSha256::parse_hex(&"ab".repeat(32)).unwrap(); + let upper = SourceManifestSha256::parse_hex(&"AB".repeat(32)).unwrap(); + assert_eq!(lower, upper, "hex casing is a rendering, not a value"); + assert_eq!( + lower.to_hex(), + "ab".repeat(32), + "canonical form is lowercase" + ); + + let a = ImportIdentity::new(lineage(), bake(), lower, WriterAbi(1)).import_id(); + let b = ImportIdentity::new(lineage(), bake(), upper, WriterAbi(1)).import_id(); + assert_eq!(a, b); + + // Two-sided: a genuinely different digest must differ. + let other = SourceManifestSha256::parse_hex(&"cd".repeat(32)).unwrap(); + assert_ne!( + a, + ImportIdentity::new(lineage(), bake(), other, WriterAbi(1)).import_id(), + "a different manifest is a different import" + ); + + // And malformed input is refused rather than silently truncated. + assert!(SourceManifestSha256::parse_hex("abc").is_err()); + assert!(SourceManifestSha256::parse_hex(&"z".repeat(64)).is_err()); + } + + // ── Semantic identity vs access residue ─────────────────────────────── + + /// PROBE (Phase A). Reopening, appending, compacting — every one of these + /// moves the head. None may change the import identity, because the + /// identity is derived from the SOURCE, not from the dataset's physical + /// state. + /// + /// System falsifier still owed (Phase C): run a real import, compact the + /// dataset, and assert the seal on disk is unchanged. + #[test] + fn access_residue_does_not_change_semantic_identity() { + let id = identity().import_id(); + let again = identity().import_id(); + assert_eq!(id, again, "the import identity is a property of the SOURCE"); + + let other = ImportIdentity::new( + lineage(), + SourceBakeId::parse("berlin-260901").unwrap(), + manifest(), + WriterAbi(1), + ) + .import_id(); + assert_ne!(id, other, "a different bake is a different import"); + } + + /// PROBE (Phase A). Concatenating variable-length fields without a + /// separator makes `("ab","c")` and `("a","bc")` collide. + #[test] + fn import_id_fields_cannot_be_confused_across_a_boundary() { + let a = ImportIdentity::new( + LineageId::parse("ab", LAYOUT_V2).unwrap(), + SourceBakeId::parse("c").unwrap(), + manifest(), + WriterAbi(1), + ) + .import_id(); + let b = ImportIdentity::new( + LineageId::parse("a", LAYOUT_V2).unwrap(), + SourceBakeId::parse("bc").unwrap(), + manifest(), + WriterAbi(1), + ) + .import_id(); + assert_ne!(a, b, "field boundaries must be part of the hash"); + } + + /// PROBE (Phase A). The marker narrows; the TUPLE decides. A marker match + /// with a different tuple is a permanent conflict, never a reconcile — + /// reconciling there would adopt another import's data as this one's. + #[test] + fn a_marker_match_with_a_different_tuple_is_a_conflict_not_a_reconcile() { + let same = reconcile(&identity(), &identity()); + assert_eq!(same, ReconcileVerdict::SameImport); + + // A genuinely different import: different marker, so not a retry. + let different = ImportIdentity::new( + lineage(), + SourceBakeId::parse("berlin-260901").unwrap(), + manifest(), + WriterAbi(1), + ); + assert_eq!( + reconcile(&identity(), &different), + ReconcileVerdict::DifferentImport + ); + + // Forced collision: same marker, different tuple. Constructed by + // asserting the marker rather than searching for a real FNV collision + // — what is under test is the DECISION, not FNV's collision rate. + let candidate = identity(); + let mut found = identity(); + found.writer = WriterAbi(2); + let verdict = if candidate.import_id() == found.import_id() { + reconcile(&candidate, &found) + } else { + // Simulate the collision the real search would surface. + ReconcileVerdict::HashConflict { + marker: candidate.import_id(), + candidate: Box::new(candidate.clone()), + found: Box::new(found.clone()), + } + }; + assert!( + matches!(verdict, ReconcileVerdict::HashConflict { .. }), + "same marker + different tuple must be permanent, not reconcilable" + ); + } + + // ── Publication vs observation ──────────────────────────────────────── + + /// PROBE (Phase A). Upstream's exact scenario: retrying while the head + /// stands at V5 must not record V5 as this import's publication position. + /// + /// System falsifier still owed (Phase C): assert `Dataset::versions()` + /// gained no entry across a reconciled retry. Constructing a `Reconciled` + /// value here proves the decision, not the absence of a second version. + #[test] + fn a_reconciled_retry_records_no_publication_version() { + let fresh = ImportSeal::seal(identity(), PublicationOutcome::Committed(v(1))) + .expect("a fresh commit seals"); + assert_eq!(fresh.publication_version, Some(v(1))); + + let retried = ImportSeal::seal( + identity(), + PublicationOutcome::Reconciled { current_head: v(5) }, + ) + .expect("a reconciled retry still seals"); + assert_eq!( + retried.publication_version, None, + "V5 is the head at reconciliation time, NOT this import's publication position" + ); + assert_eq!( + retried.observed_head, + Some(v(5)), + "the observed head is a read horizon and is recorded as one" + ); + assert_eq!( + retried.import_id, fresh.import_id, + "the durable identity is the same across the retry" + ); + } + + /// PROBE (Phase A). An ambiguous publication seals NOTHING — the identity + /// is re-submitted frozen and reconciled. There is no compensating delete. + /// + /// System falsifier still owed (Phase C/D): kill the process between + /// commit and acknowledgement on a first create, restart, re-submit, and + /// assert exactly one version exists. + #[test] + fn an_ambiguous_publication_seals_nothing() { + assert!( + ImportSeal::seal(identity(), PublicationOutcome::Ambiguous).is_none(), + "an unknown outcome must not be recorded as a success" + ); + } + + // ── Head decisions ──────────────────────────────────────────────────── + + /// PROBE (Phase A). A later unexplained head is drift: serve the clean + /// checkout, report it, keep the lineage. + #[test] + fn an_unexplained_later_head_serves_the_clean_checkout_and_reports_drift() { + let none = KnownMaintenance::none(); + let have = versions(&[7, 9]); + let d = decide_head(&HeadFacts { + clean: v(7), + observed_head: v(9), + known_maintenance: &none, + locally_available: &have, + }); + assert_eq!( + d, + HeadDecision::CheckoutClean { + clean: v(7), + head: v(9) + } + ); + assert!(d.head_drift()); + assert_eq!( + d.serving(), + Some(ServingVersion { + version: v(7), + pinned_to_clean: true + }), + "serve the last known clean sealed version, not the mystery head" + ); + + // Silence half: a recognised maintenance descendant is NOT drift, or + // every compaction would pin the replica backwards forever. + let known = KnownMaintenance::asserted(versions(&[9])); + let ok = decide_head(&HeadFacts { + clean: v(7), + observed_head: v(9), + known_maintenance: &known, + locally_available: &have, + }); + assert_eq!( + ok, + HeadDecision::ServeKnownDescendant { + version: v(9), + provenance: MaintenanceProvenance::CallerAsserted + }, + "the decision carries WHERE the recognition came from — Phase A can \ + only assert it, and the status endpoint must be able to say so" + ); + assert!(!ok.head_drift()); + + // And an unmoved head is plain serving. + assert_eq!( + decide_head(&HeadFacts { + clean: v(7), + observed_head: v(7), + known_maintenance: &none, + locally_available: &have, + }), + HeadDecision::ServeHead(v(7)) + ); + } + + /// PROBE (Phase A). A head that is *behind* the clean import is not drift + /// — it is a truncated replica, and asking it to check out a version it + /// may not possess is a different bug from serving a mystery head. + #[test] + fn a_head_behind_clean_is_a_replica_problem_not_a_drift() { + let none = KnownMaintenance::none(); + let have = versions(&[3]); + let d = decide_head(&HeadFacts { + clean: v(7), + observed_head: v(3), + known_maintenance: &none, + locally_available: &have, + }); + assert_eq!( + d, + HeadDecision::ReplicaIncomplete { + clean: v(7), + head: v(3), + reason: BehindReason::HeadBehindClean + } + ); + assert!(!d.head_drift(), "behind is not drift"); + assert_eq!( + d.serving(), + None, + "there is nothing to serve until recovery runs — a checkout would \ + request a version this replica does not have" + ); + } + + /// PROBE (Phase A). The clean version being absent locally is the same + /// recovery case, reached from the other direction. + #[test] + fn an_absent_clean_version_is_never_offered_as_a_checkout() { + let none = KnownMaintenance::none(); + let have = versions(&[9]); // head present, clean NOT + assert_eq!( + decide_head(&HeadFacts { + clean: v(7), + observed_head: v(9), + known_maintenance: &none, + locally_available: &have, + }), + HeadDecision::ReplicaIncomplete { + clean: v(7), + head: v(9), + reason: BehindReason::CleanVersionAbsent + } + ); + } + + /// PROBE (Phase A). Recovery prefers hydrating the intended version; + /// falling back to an older clean one is a real degradation and is only + /// chosen when S3 cannot help. + #[test] + fn recovery_hydrates_when_it_can_and_degrades_only_when_it_must() { + let available = versions(&[3, 5]); + assert_eq!( + recover_behind(v(7), &available, true), + BehindRecovery::HydrateFromS3 { want: v(7) } + ); + assert_eq!( + recover_behind(v(7), &available, false), + BehindRecovery::ServeAvailablePrevious { version: v(5) }, + "the NEWEST available clean version below the target, not the oldest" + ); + assert_eq!( + recover_behind(v(7), &BTreeSet::new(), false), + BehindRecovery::Unserviceable, + "nothing serviceable is reported, not faked" + ); + } + + // ── Lifecycle: serving × operation ──────────────────────────────────── + + /// PROBE (Phase A). Rotation happens on activation and nowhere else. + #[test] + fn activation_rotates_active_to_previous() { + let a = ImportId(0xA); + let b = ImportId(0xB); + let l = Lifecycle::serving(a, None); + let (l, g) = l.start_import(b); + assert_eq!( + l.active(), + Some(a), + "an import in flight does not disturb what is being served" + ); + let l = l.activate(b, g).expect("the fence matches"); + assert_eq!(l.active(), Some(b)); + assert_eq!(l.previous(), Some(a)); + assert_eq!(l.operation, Operation::Idle); + } + + /// PROBE (Phase A). The rule the product state exists to make structural. + #[test] + fn a_failed_candidate_never_disturbs_active_or_previous() { + let a = ImportId(0xA); + let p = ImportId(0x9); + let bad = ImportId(0xBAD); + let before = Lifecycle::serving(a, Some(p)); + + let (l, g) = before.start_import(bad); + let l = l + .fail(bad, g, ImportFailure::Validation("row count".into())) + .expect("the fence matches"); + + assert_eq!( + l.serving, before.serving, + "both the active AND the previous-clean artifact survive a failed candidate" + ); + assert!(matches!(l.operation, Operation::Failed { .. })); + + // ...and recovery from that state still rotates correctly. + let c = ImportId(0xC); + let (l, g) = l.start_import(c); + let l = l.activate(c, g).unwrap(); + assert_eq!(l.active(), Some(c)); + assert_eq!(l.previous(), Some(a)); + } + + /// PROBE (Phase A). The fence. B's late completion must not activate or + /// fail C — this is the concurrency bug the generation exists to stop, and + /// without the fence B's success would install B while C was mid-build. + #[test] + fn a_stale_completion_cannot_activate_or_fail_the_superseding_candidate() { + let a = ImportId(0xA); + let b = ImportId(0xB); + let c = ImportId(0xC); + + let (l, gen_b) = Lifecycle::serving(a, None).start_import(b); + let (l, gen_c) = l.start_import(c); + assert_ne!(gen_b, gen_c, "starting C must advance the fence"); + + let stale = l.activate(b, gen_b); + assert_eq!( + stale, + Err(LifecycleError::StaleCompletion { + candidate: b, + generation: gen_b, + current: Some(gen_c) + }), + "B's late success must not install B over the in-flight C" + ); + let stale_fail = l.fail(b, gen_b, ImportFailure::Write("boom".into())); + assert!( + matches!(stale_fail, Err(LifecycleError::StaleCompletion { .. })), + "nor may B's late failure mark C as failed" + ); + + // Can-it-succeed half: C's own completion at its own generation works. + let l = l.activate(c, gen_c).expect("C's completion is current"); + assert_eq!(l.active(), Some(c)); + assert_eq!(l.previous(), Some(a)); + } + + /// PROBE (Phase A). A completion with nothing in flight is refused rather + /// than silently applied. + #[test] + fn a_completion_with_no_import_in_flight_is_refused() { + let l = Lifecycle::serving(ImportId(0xA), None); + assert_eq!( + l.activate(ImportId(0xB), Generation(1)), + Err(LifecycleError::NoImportInFlight { + candidate: ImportId(0xB) + }) + ); + } + + /// PROBE (Phase A). First create: nothing was active, so there is no + /// previous to invent. + #[test] + fn the_first_activation_has_no_previous() { + let (l, g) = Lifecycle::absent().start_import(ImportId(1)); + assert_eq!( + l.active(), + None, + "nothing is served during the first import" + ); + let l = l.activate(ImportId(1), g).unwrap(); + assert_eq!(l.active(), Some(ImportId(1))); + assert_eq!(l.previous(), None); + } + + // ── Activation gate ─────────────────────────────────────────────────── + + #[test] + fn an_ephemeral_candidate_waits_for_s3_and_a_volume_does_not() { + let p = ActivationPolicy::default(); + assert_eq!( + decide_activation(RootKind::Ephemeral, false, p), + Activation::HoldForS3, + "an un-mirrored ephemeral candidate would be re-converted on the next deploy" + ); + assert_eq!( + decide_activation(RootKind::Volume, false, p), + Activation::ActivateS3Lagging, + "a volume survives a redeploy, so it may run ahead of the mirror" + ); + // Silence half: with S3 in hand both activate plainly. + assert_eq!( + decide_activation(RootKind::Ephemeral, true, p), + Activation::Activate + ); + assert_eq!( + decide_activation(RootKind::Volume, true, p), + Activation::Activate + ); + } + + // ── Storage policy ──────────────────────────────────────────────────── + + #[test] + fn the_storage_policy_is_ordered_and_falls_back_to_ephemeral() { + let full = RootInputs { + osm_artifact_root: Some("/a".into()), + osm_slab_cache_dir: Some("/b".into()), + railway_vol: Some("/c".into()), + volume01_writable: true, + ephemeral_dir: PathBuf::from("/tmp/osm"), + ..RootInputs::default() + }; + assert_eq!(resolve_root(&full).path, PathBuf::from("/a")); + + let no_first = RootInputs { + osm_artifact_root: None, + ..full.clone() + }; + assert_eq!(resolve_root(&no_first).path, PathBuf::from("/b")); + + let no_two = RootInputs { + osm_slab_cache_dir: None, + ..no_first.clone() + }; + assert_eq!(resolve_root(&no_two).path, PathBuf::from("/c/osm")); + + let no_three = RootInputs { + railway_vol: None, + ..no_two.clone() + }; + assert_eq!(resolve_root(&no_three).path, PathBuf::from("/volume01/osm")); + + let none = RootInputs { + volume01_writable: false, + ..no_three.clone() + }; + let r = resolve_root(&none); + assert_eq!(r.path, PathBuf::from("/tmp/osm")); + assert_eq!( + r.kind, + RootKind::Ephemeral, + "the fallback must be REPORTED as ephemeral" + ); + + // An empty variable is an absent one — a Railway row created and never + // filled must not resolve to "". + let blank = RootInputs { + osm_artifact_root: Some(" ".into()), + ..full.clone() + }; + assert_eq!(resolve_root(&blank).path, PathBuf::from("/b")); + + // ...and so is a RELATIVE one: it would resolve against whatever + // directory the server was launched from. + let relative = RootInputs { + osm_artifact_root: Some("osm-data".into()), + ..full + }; + assert_eq!(resolve_root(&relative).path, PathBuf::from("/b")); + } + + /// PROBE (Phase A). A named path is not a durable path. + /// + /// `OSM_ARTIFACT_ROOT` and `OSM_SLAB_CACHE_DIR` routinely point at + /// container-local scratch. Classifying them as `Volume` because someone + /// set them would let an ephemeral candidate activate without S3 — and the + /// next redeploy would re-run the whole SOA→Lance conversion, which is the + /// exact cost this design exists to remove. + #[test] + fn a_generic_override_is_ephemeral_unless_durability_is_declared() { + let undeclared = RootInputs { + osm_artifact_root: Some("/scratch/osm".into()), + ..RootInputs::default() + }; + let r = resolve_root(&undeclared); + assert_eq!(r.path, PathBuf::from("/scratch/osm")); + assert_eq!( + r.kind, + RootKind::Ephemeral, + "an undeclared override must default to ephemeral — the fail-safe direction" + ); + assert_eq!( + decide_activation(r.kind, false, ActivationPolicy::default()), + Activation::HoldForS3, + "and that classification must actually gate activation" + ); + + // Can-it-be-durable half: an explicit declaration is honoured. + let declared = RootInputs { + override_durability: Some(RootKind::Volume), + ..undeclared + }; + assert_eq!(resolve_root(&declared).kind, RootKind::Volume); + + // RAILWAY_VOL is intrinsically a volume — no declaration needed. + let railway = RootInputs { + railway_vol: Some("/data".into()), + ..RootInputs::default() + }; + let r = resolve_root(&railway); + assert_eq!(r.path, PathBuf::from("/data/osm")); + assert_eq!(r.kind, RootKind::Volume); + } + + /// PROBE (Phase A). A defaulted `RootInputs` must not write the artifact + /// tree into the process's current directory. + #[test] + fn the_default_root_is_absolute_and_never_the_current_directory() { + let r = resolve_root(&RootInputs::default()); + assert!( + r.path.is_absolute(), + "the default resolved to {:?}, which resolves against the cwd", + r.path + ); + assert!(!r.path.as_os_str().is_empty()); + assert_eq!(r.kind, RootKind::Ephemeral); + + // A hand-built empty/relative fallback is corrected, not honoured. + for bad in ["", "relative/osm"] { + let inputs = RootInputs { + ephemeral_dir: PathBuf::from(bad), + ..RootInputs::default() + }; + let r = resolve_root(&inputs); + assert!( + r.path.is_absolute(), + "ephemeral_dir {bad:?} resolved to {:?}", + r.path + ); + } + } + + // ── Publication order ───────────────────────────────────────────────── + + /// PROBE (Phase A). System falsifier still owed (Phase D): observe the + /// real S3 write order under a killed process. + #[test] + fn the_pointer_is_published_last_and_present_objects_are_not_re_uploaded() { + let already: BTreeSet = ["frag/1".to_string()].into_iter().collect(); + let steps = plan_publication( + ImportId(0xB), + Some(ImportId(0xA)), + &["frag/1".to_string(), "frag/2".to_string()], + &already, + &["import-seals/000000000000000b.yaml".to_string()], + ); + + assert!( + matches!(steps.first(), Some(PublishStep::EnsurePreviousDurable(id)) if *id == ImportId(0xA)) + ); + assert!( + matches!(steps.last(), Some(PublishStep::PublishPointer { .. })), + "the head pointer must be the LAST step — a reader must never see it before the data" + ); + match &steps[1] { + PublishStep::UploadMissingObjects(m) => assert_eq!( + m, + &vec!["frag/2".to_string()], + "a fragment already referenced in S3 must never be uploaded twice" + ), + other => panic!("expected the upload step, got {other:?}"), + } + } + + // ── Warm boot ───────────────────────────────────────────────────────── + + /// PROBE (Phase A). This proves the READ PLAN names no data object. It + /// does **not** prove a real boot performs zero data traversal — the boot + /// path does not exist yet. + /// + /// System falsifier still owed (Phase B/C): instrument the real warm-boot + /// path and assert the bytes read are bounded by the seals' size. + #[test] + fn the_warm_boot_read_plan_names_only_small_seals() { + let reads = warm_verification_reads(&lineage(), ImportId(0xB)); + assert_eq!(reads.len(), 3); + for key in &reads { + assert!( + key.ends_with(".yaml"), + "warm verification named {key}, which is not a seal — proving an origin \ + must never require reading map data" + ); + assert!( + !key.contains("/data/") && !key.contains(".soa") && !key.contains(".lance/data"), + "warm verification must not name a data object: {key}" + ); + } + assert!(reads.iter().any(|k| k.ends_with("origin-seal.yaml"))); + assert!( + reads + .iter() + .any(|k| k.contains("import-seals/000000000000000b.yaml")) + ); + } + + // ── Origin seal ─────────────────────────────────────────────────────── + + /// PROBE (Phase A). The seal cannot contradict itself, and it identifies + /// the import that made it — without `origin_import_id` the first sealed + /// import is unfindable. + #[test] + fn an_origin_seal_is_internally_consistent_and_names_its_import() { + let seal = OriginSeal::create(identity(), Some(v(1))); + assert_eq!( + seal.origin_import_id(), + identity().import_id(), + "the origin import id is DERIVED — it cannot disagree with the identity" + ); + assert_eq!( + seal.layout(), + LAYOUT_V2, + "the layout is read from the lineage; there is no second copy to drift" + ); + assert_eq!(seal.initial_bake(), &bake()); + assert_eq!(seal.initial_manifest(), &manifest()); + assert_eq!(seal.lineage(), &lineage()); + + // A later import in the same lineage is accepted; another lineage is not. + let later = ImportIdentity::new( + lineage(), + SourceBakeId::parse("berlin-260901").unwrap(), + manifest(), + WriterAbi(1), + ); + assert!(seal.accepts(&later), "a new bake stays in the same lineage"); + let v3 = ImportIdentity::new( + LineageId::parse("berlin", LayoutAbi(3)).unwrap(), + bake(), + manifest(), + WriterAbi(1), + ); + assert!( + !seal.accepts(&v3), + "an incompatible layout is a new lineage" + ); + } + + // ── Retention ───────────────────────────────────────────────────────── + + /// PROBE (Phase A). Retention keeps; it does not delete. + /// + /// The keep-set is the union of two **reachability closures** plus every + /// seal. Deletion needs three gates open at once, and in Phase A the + /// authority gate is shut by construction. + #[test] + fn retention_keeps_everything_reachable_and_authorises_no_deletion_in_phase_a() { + let active = VersionInventory::complete( + v(9), + ["b/manifest", "b/frag1", "b/idx", "b/del1", "b/txn"] + .into_iter() + .map(String::from) + .collect(), + ); + let previous = VersionInventory::complete( + v(7), + ["a/manifest", "a/frag1"] + .into_iter() + .map(String::from) + .collect(), + ); + let seals: BTreeSet = ["s/a.yaml", "s/b.yaml", "s/old.yaml"] + .into_iter() + .map(String::from) + .collect(); + let everything: BTreeSet = [ + "b/manifest", + "b/frag1", + "b/idx", + "b/del1", + "b/txn", + "a/manifest", + "a/frag1", + "s/a.yaml", + "s/b.yaml", + "s/old.yaml", + "ancient/1", + "ancient/2", + ] + .into_iter() + .map(String::from) + .collect(); + + // Phase A: grace elapsed, inventories complete — and STILL no deletion. + let p = plan_retention( + &active, + Some(&previous), + &seals, + &everything, + true, + GcAuthority::None, + ); + assert_eq!( + p, + RetentionPlan::KeepOnly { + keep: p.keep().clone(), + hold: RetentionHold::NoGcAuthority + }, + "without a Lance-native reachability mechanism, `everything - keep` is a \ + guess dressed as a set operation" + ); + assert!(p.collect().is_empty()); + assert!( + p.keep().contains("b/del1"), + "deletion files are reachable state" + ); + assert!(p.keep().contains("b/idx"), "indices are reachable state"); + assert!( + p.keep().contains("a/manifest"), + "previous-clean is kept whole" + ); + assert!( + p.keep().contains("s/old.yaml"), + "every seal is kept — a seal is the only record of what a version meant" + ); + + // Phase D shape: with authority AND a complete inventory AND the grace + // period elapsed, only genuinely unreferenced objects are collectable. + let d = plan_retention( + &active, + Some(&previous), + &seals, + &everything, + true, + GcAuthority::LanceNativeReachability, + ); + assert_eq!( + d.collect(), + ["ancient/1".to_string(), "ancient/2".to_string()] + .into_iter() + .collect() + ); + let remaining: BTreeSet<_> = everything.difference(&d.collect()).cloned().collect(); + assert_eq!(&remaining, d.keep(), "bounded: active + previous + seals"); + + // Gate 1: the grace period must actually hold. + assert_eq!( + plan_retention( + &active, + Some(&previous), + &seals, + &everything, + false, + GcAuthority::LanceNativeReachability, + ), + RetentionPlan::KeepOnly { + keep: d.keep().clone(), + hold: RetentionHold::GracePeriodActive + } + ); + } + + /// PROBE (Phase A). The invariant that matters most: an incomplete + /// inventory produces **no deletion plan at all**, even with full + /// authority and the grace period long past. A partial closure that gets + /// differenced against "everything" deletes exactly the objects it failed + /// to enumerate. + #[test] + fn an_incomplete_inventory_produces_no_deletion_plan() { + let everything: BTreeSet = ["x", "y", "z"].into_iter().map(String::from).collect(); + let partial = + VersionInventory::partial(v(9), ["x"].into_iter().map(String::from).collect()); + let complete = + VersionInventory::complete(v(7), ["y"].into_iter().map(String::from).collect()); + + for (a, p) in [(&partial, Some(&complete)), (&complete, Some(&partial))] { + let plan = plan_retention( + a, + p, + &BTreeSet::new(), + &everything, + true, + GcAuthority::LanceNativeReachability, + ); + assert!( + matches!( + plan, + RetentionPlan::KeepOnly { + hold: RetentionHold::IncompleteInventory, + .. + } + ), + "an incomplete inventory must never authorise deletion, got {plan:?}" + ); + assert!(plan.collect().is_empty()); + } + + // Can-it-ever-collect half, so the guard is not simply always-on. + let ok = plan_retention( + &complete, + None, + &BTreeSet::new(), + &everything, + true, + GcAuthority::LanceNativeReachability, + ); + assert_eq!( + ok.collect(), + ["x".to_string(), "z".to_string()].into_iter().collect() + ); + } + + // ── Lineage identity ────────────────────────────────────────────────── + + #[test] + fn an_incompatible_layout_is_a_different_lineage() { + let v2 = LineageId::parse("berlin", LAYOUT_V2).unwrap(); + let v3 = LineageId::parse("berlin", LayoutAbi(3)).unwrap(); + assert_ne!(v2, v3); + assert_eq!(v2.path_segment(), "berlin/osm-lance-v2"); + assert_eq!(v3.path_segment(), "berlin/osm-lance-v3"); + assert!( + !v2.accepts_layout(LayoutAbi(3)), + "a v3 dataset is not servable as v2" + ); + assert!(v2.accepts_layout(LAYOUT_V2)); + + // A new BAKE at the same ABI stays in the same lineage — the property + // that stops every import minting a new S3 prefix. + let a = ImportIdentity::new( + v2.clone(), + SourceBakeId::parse("aug").unwrap(), + manifest(), + WriterAbi(1), + ) + .import_id(); + let b = ImportIdentity::new( + v2.clone(), + SourceBakeId::parse("sep").unwrap(), + manifest(), + WriterAbi(1), + ) + .import_id(); + assert_ne!(a, b, "different bakes are different imports"); + assert_eq!( + v2.path_segment(), + "berlin/osm-lance-v2", + "...in ONE lineage" + ); + } + + /// Two regions never share a lineage even at the same ABI. + #[test] + fn regions_have_independent_lineages() { + let berlin = LineageId::parse("berlin", LAYOUT_V2).unwrap(); + let bb = LineageId::parse("brandenburg", LAYOUT_V2).unwrap(); + assert_ne!(berlin.path_segment(), bb.path_segment()); + assert_ne!( + ImportIdentity::new(berlin, bake(), manifest(), WriterAbi(1)).import_id(), + ImportIdentity::new(bb, bake(), manifest(), WriterAbi(1)).import_id(), + "the same bake into two regions is two imports" + ); + } + + /// The writer ABI participates in the identity: the same source through an + /// incompatible writer is a different import, not a reconcilable retry. + #[test] + fn the_writer_abi_participates_in_the_import_identity() { + assert_ne!( + ImportIdentity::new(lineage(), bake(), manifest(), WriterAbi(1)).import_id(), + ImportIdentity::new(lineage(), bake(), manifest(), WriterAbi(2)).import_id(), + ); + } +} diff --git a/crates/stubs/notebook-query/Cargo.toml b/crates/stubs/notebook-query/Cargo.toml index 2bbd8ef81..5e6086a2f 100644 --- a/crates/stubs/notebook-query/Cargo.toml +++ b/crates/stubs/notebook-query/Cargo.toml @@ -8,11 +8,21 @@ edition = "2024" lance-graph.workspace = true lance-graph-contract.workspace = true bgz17.workspace = true -# Arrow + DataFusion versions MUST track lance-graph (arrow 58, datafusion 53), -# else `CypherQuery::execute` returns an arrow-58 `RecordBatch` that won't unify -# with a locally-built arrow-57 one (E0308). See lance-graph/crates/lance-graph. +# Arrow version MUST track lance-graph (arrow 58), else `CypherQuery::execute` +# returns an arrow-58 `RecordBatch` that won't unify with a locally-built +# arrow-57 one (E0308). See lance-graph/crates/lance-graph. +# +# `datafusion` itself is NOT a direct dependency here — nothing in this crate +# references it (verified: zero `datafusion::` uses in src/). A stale direct +# pin at "53" used to sit here "for version alignment" while lance-graph had +# already moved to the lance-9 family (datafusion 54, lancedb 0.33, arrow 58); +# an unused pin at the WRONG version is strictly worse than no pin — it forces +# cargo to build an entire second datafusion-53 + arrow-57 dependency subtree +# alongside the one lance-graph already needs, for nothing. Removed rather +# than bumped to "54": if this crate ever needs `datafusion::` types directly, +# add it back tracking whatever lance-graph pins at that time, not a hardcoded +# major version that can drift again. arrow = { version = "58", features = ["prettyprint"] } -datafusion = { version = "53", default-features = false } tokio = { version = "1", features = ["rt-multi-thread", "macros"] } serde.workspace = true serde_json = "1"