diff --git a/config/production.toml b/config/production.toml index 72e82112..2fd9a3df 100644 --- a/config/production.toml +++ b/config/production.toml @@ -49,57 +49,18 @@ verify_on_read = true # Maximum size in GiB of the legacy LMDB store, while one still exists # (0 = derive it from available disk). Retired along with LMDB itself. db_size_gb = 0 - -# --- Moving off the legacy LMDB chunk store --- +# Storage migration off the old chunk store # # Chunks are now one file each, under {root_dir}/chunks/. A node that still has a # chunks.mdb copies it into files in the background, then deletes it whole, which is the # only moment LMDB's disk comes back. # -# The two release-level switches (whether to delete the old store, and whether audits -# still penalise) belong to the build, not to this file, so they are deliberately absent. -[storage.migration] -# Run the copier. Turning this off leaves both stores in place forever and never -# returns the old store's disk. -enabled = true - -# Also write new chunks to the legacy store while it exists, so a fleet rollback to an -# older build cannot lose a chunk uploaded during the migration. -dual_write_legacy = true - -# Allow a node that cannot fit its chunks to give up the ones it is furthest from. -# -# Whatever this is set to, a chunk is only ever given up when the node is near the back of -# its group for it, its close group has received the node's reduced commitment, AND all but -# one of that group has cryptographically proven it holds a copy. A node that cannot show -# all three keeps both stores and asks for more disk. Turn this off if you would rather add -# disk than have the node give anything up at all. -allow_shed = true - -# Hours after this build first starts before a node may give anything up, so peers on -# older builds have upgraded and stopped penalising it for doing so. -shed_hold_hours = 72 - -# Hours between one migration wave opening and the next. -# -# A close group is split into waves so only two of its members give chunks up at a time. -# If all seven went together none could prove to the others that a copy survived, and the -# group would deadlock waiting on each other. A node with room to copy everything does not -# wait for a wave: it is never unable to serve, so it is not part of that problem. -wave_hours = 24 - -# Hours between a node committing to what it will keep and deleting the old store. -# Never shorter than 4: that is what the answerability window needs. -retire_delay_hours = 4 - -# Free space, in MiB, the copier leaves untouched on top of disk_reserve_mb. -copier_slack_mb = 2048 - -# Copy rate ceiling, in MiB/s. Keep it modest: an unthrottled copier competing with the -# audit responder for disk turns a storage migration into an audit incident. -copier_throttle_mib_per_sec = 32 +# There is deliberately no [storage.migration] section. Every one of its settings belongs +# to the build rather than to this file: the release that later deletes the old store has +# to be able to assume every node ran the same schedule, and it cannot assume that while +# the schedule belongs to whoever last edited a config. A [storage.migration] section left +# over from an earlier release still parses, it is simply ignored. -# --- Upgrade --- [upgrade] enabled = false channel = "stable" diff --git a/deploy/terraform/cloud-init/worker.yml b/deploy/terraform/cloud-init/worker.yml index 391fe761..49e6684c 100644 --- a/deploy/terraform/cloud-init/worker.yml +++ b/deploy/terraform/cloud-init/worker.yml @@ -79,7 +79,12 @@ write_files: Type=simple User=ant Group=ant - ExecStart=$${BINARY_PATH} --root-dir $${NODE_DIR} --port 0 --metrics-port $${METRICS_PORT} $${BOOTSTRAP_ARGS} + # --enable-logging is not optional here, whatever it looks like. Without it the + # binary installs no tracing subscriber and emits nothing at all, so a node on this + # unit is silent: no migration progress, no warnings, no way to tell a fleet that has + # finished moving off the old chunk store from one that has not. JSON because what + # reads these lines is a query, and the fields that matter are structured. + ExecStart=$${BINARY_PATH} --root-dir $${NODE_DIR} --port 0 --metrics-port $${METRICS_PORT} --enable-logging --log-format json $${BOOTSTRAP_ARGS} Restart=always RestartSec=10 MemoryMax=350M diff --git a/docs/adr/ADR-0014-file-based-chunk-store-and-lmdb-retirement.md b/docs/adr/ADR-0014-file-based-chunk-store-and-lmdb-retirement.md index c0631a2f..1518679f 100644 --- a/docs/adr/ADR-0014-file-based-chunk-store-and-lmdb-retirement.md +++ b/docs/adr/ADR-0014-file-based-chunk-store-and-lmdb-retirement.md @@ -370,6 +370,49 @@ belief carry its own expiry — the directory carries its mark, the proof carrie saw, the write carries its note — rather than to check again and hope the check is close enough to the act. +## Amendment: the schedule belongs to the build + +Shipped after this record's release, as a patch on top of it. + +`MigrationConfig` is now `serde(skip)` on every field. A `[storage.migration]` section left in +an operator's file still parses and is ignored. The release that later deletes the old store +has to be able to assume every node ran the same schedule, and it cannot assume that while the +schedule belongs to whoever last edited a config. `wave_hours` goes from 24 to 12 so the whole +schedule fits inside the window before that release. + +Two behaviour changes follow from it and are intended, not accidents: + +- A node whose file says `enabled = false` migrates anyway. Nobody may opt out. +- A node an operator had paused resumes with its hold and its waves already spent, because the + marker is stamped when the store opens rather than when the copier starts. Nothing + distinguishes it from a node whose disk filled on day one, and restamping both would add + three days of waiting to the nodes with the least room. What is lost is the stagger, not the + safety: a chunk is still only given up once all but one of the close group has proven it + holds a copy. Pinned by a test. + +Halving the wave spacing only ever moves a node earlier. Both schedules share the same base and +the same wave index and only the multiplier shrinks, so no wave that had opened closes again. +While the fleet is mixed the two schedules can put different waves in the same slot, which +costs churn and not data. + +The one path that deleted a chunk without a possession check is gone. When the file store +refused a legacy value for exceeding the per-chunk ceiling, the copier deleted it outright. +That case cannot occur — `MAX_CHUNK_SIZE` has been 4 MiB since ant-protocol's first commit, +the chunk store and its size check arrived together in v0.4.0, replication arrived already +checking on receive and fetch, and `FileStore::put` checks the address before the size — but a +branch that destroys a chunk it cannot replace does not get to rely on being unreachable. The +generic error path now handles it: refuse, name the key and the size, release the volume lock, +retry. Pinned by a test, because an impossible case is the one a later reader tidies away. + +`storage_empty` in the commitment rotation now means no bytes anywhere, not "nothing left to +commit to". Two nodes reported an empty set while holding data: one that could not copy +anything before its disk filled, whose keys are all still in the legacy store, and one whose +last readable file has gone transiently bad, since `all_keys` drops anything marked suspect or +known-wrong. Both took the `clear_all` branch and dropped every retained root, and an auditor +holding a valid pin then got `UnknownCommitment`, graded as a confirmed failure, on a node that +could have answered. Emptiness is now asked of `current_chunks`, the union of the raw file +index and the legacy-only set, and a read error reads as not-empty. + ## Consequences ### Positive diff --git a/src/config.rs b/src/config.rs index 5f11a49d..c9e9b9fe 100644 --- a/src/config.rs +++ b/src/config.rs @@ -613,25 +613,50 @@ mod tests { #[test] fn the_shipped_storage_config_parses() { - // The migration section is operator-facing, so a typo in it would only surface on - // a node that had already shipped. Only `[storage]` is checked: the rest of - // `production.toml` does not currently deserialize as a `NodeConfig` (its - // `evm_network` is a bare string where an internally tagged enum is expected), - // which is a separate, pre-existing problem. + // Only `[storage]` is checked: the rest of `production.toml` does not currently + // deserialize as a `NodeConfig` (its `evm_network` is a bare string where an + // internally tagged enum is expected), which is a separate, pre-existing problem. let raw = include_str!("../config/production.toml"); let doc: toml::Value = toml::from_str(raw).expect("production.toml must be valid TOML"); let storage = doc.get("storage").expect("a [storage] section").clone(); let config: StorageConfig = storage.try_into().expect("[storage] must deserialize"); - assert!(config.migration.enabled); - assert!(config.migration.dual_write_legacy); - assert_eq!(config.migration.shed_hold_hours, 72); - assert_eq!(config.migration.copier_throttle_mib_per_sec, 32); - assert_eq!(config.migration.copier_slack_mb, 2048); - // The release switches are absent from the file on purpose, so they come from the - // build rather than from whatever an operator's config last recorded. - let build = MigrationConfig::default(); - assert_eq!(config.migration.retire_legacy, build.retire_legacy); + // Every migration setting comes from the build. The shipped file no longer carries + // a `[storage.migration]` section at all, and would be ignored if it did. + assert_eq!(config.migration, MigrationConfig::default()); + } + + /// A config file cannot put a node on a schedule of its own. + /// + /// This is the whole guarantee the release that deletes the old chunk store rests on: + /// it may assume every node ran the same migration schedule, and it may only assume + /// that while no file can change one. The obvious way to break this is to give one of + /// these fields a `serde(default)` again, which looks like a tidy-up and is not, so it + /// is worth a test rather than a comment. Every value below is deliberately different + /// from the build's. + #[test] + fn a_config_file_cannot_change_the_migration_schedule() { + let hostile = r" +enabled = true +[migration] +enabled = false +dual_write_legacy = false +allow_shed = false +shed_hold_hours = 9999 +retire_delay_hours = 9999 +wave_hours = 9999 +copier_slack_mb = 1 +copier_throttle_mib_per_sec = 1 +tick_secs = 9999 +batch_chunks = 1 +"; + let config: StorageConfig = + toml::from_str(hostile).expect("an old or hostile file must still parse"); + assert_eq!( + config.migration, + MigrationConfig::default(), + "the file changed the migration schedule, which the final release cannot survive" + ); } #[test] diff --git a/src/node.rs b/src/node.rs index aa5d1fe7..10a8bd7f 100644 --- a/src/node.rs +++ b/src/node.rs @@ -317,6 +317,23 @@ impl NodeBuilder { } } + // Say on the wire whether this node still has an old chunk store. It costs no new + // message and no new field: saorsa-core already sends a user agent with every signed + // message and keeps each peer's, so this is a different value in a string that was + // already there. It is how the release that deletes the old store finds out whether + // the fleet has finished, including the nodes we do not run and have no logs from. + // + // Read from the filesystem here rather than from the store, because the store is + // built later and a node with storage switched off never builds one at all, while + // the directory on its disk is just as real either way. + // + // Fixed for the life of the process: saorsa-core copies the string when it builds + // the transport. A node that finishes migrating goes on saying `legacy` until it + // restarts, which overstates how much is left rather than understating it, and is + // the direction a release gate should err in. + let signal = crate::storage::MigrationSignal::from_disk(&config.root_dir); + core_config.custom_user_agent = Some(crate::storage::migration_signal::user_agent(signal)); + // Persist close group peers + trust scores across restarts. // Default to root_dir (alongside node_identity.key) when not explicitly set. core_config.close_group_cache_dir = Some( @@ -442,8 +459,12 @@ impl NodeBuilder { } if config.upgrade.staged_rollout_hours > 0 { - monitor = - monitor.with_staged_rollout(node_id_seed, config.upgrade.staged_rollout_hours); + monitor = monitor + .with_staged_rollout(node_id_seed, config.upgrade.staged_rollout_hours) + // Under the node's own root, not the machine-wide upgrade cache: this is one + // node's place in one window, and the cache is shared by every node on the + // host. Without it the window restarts whenever the node does. + .with_rollout_state(&config.root_dir); } monitor @@ -655,6 +676,24 @@ impl RunningNode { info!("Replication engine started"); } + // Say where this node is in the storage migration, and what it can see around it. + // + // The release that deletes the old chunk store may only be published once the fleet + // has moved, and no calendar establishes that. Our own logs cover the nodes we run; + // this covers the ones every node can see, which includes the ones we do not run and + // would otherwise have no view of at all. + { + // Weak on purpose: see `report_until_shutdown`. A reporter that kept the node + // alive would keep its port bound after the node was dropped. + let p2p = Arc::downgrade(&self.p2p_node); + let root_dir = self.config.root_dir.clone(); + let shutdown = self.shutdown.clone(); + tokio::spawn(async move { + crate::storage::migration_signal::report_until_shutdown(p2p, root_dir, shutdown) + .await; + }); + } + // Start upgrade monitor if enabled if let Some(monitor) = self.upgrade_monitor.take() { let events_tx = self.events_tx.clone(); diff --git a/src/replication/mod.rs b/src/replication/mod.rs index 71124727..d37e1c03 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -9980,7 +9980,29 @@ async fn rebuild_and_rotate_commitment( // this filter the pruner's reprieve would keep re-committing stale keys // forever (the rebuild reads all_keys, so a retained-on-disk key would be // re-committed and re-gossiped every rotation — a permanent pin). - let storage_empty = stored_keys.is_empty(); + // "Empty" has to mean this node holds no bytes at all, not that it has nothing left to + // COMMIT to. Once the migration has settled, `committable_keys` narrows to the file-backed + // set, so a node that could not copy anything before its disk filled reports an empty set + // while a full legacy environment sits beside it. Reading that as "the bytes are gone" + // takes the `clear_all` branch below and drops every retained root, and an auditor still + // holding a valid pin then gets `UnknownCommitment`, which is graded as a confirmed + // failure. That node can answer the challenge: the bytes are right there in the legacy + // store. It is the `retire_current` branch it belongs in. + // Asked of what is physically on disk, not of what this node will currently answer for. + // The two differ: `all_keys` drops a file the store has marked suspect or known-wrong, so + // a node whose last readable file has gone transiently bad has an empty committable set + // while the bytes are still there and may be readable again in a moment. So does a node + // that could not copy anything before its disk filled, whose keys are all still in the + // legacy store. Reading either as "the bytes are gone" takes the `clear_all` branch below + // and drops every retained root, and an auditor holding a valid pin then gets + // `UnknownCommitment`, graded as a confirmed failure, on a node that could have answered. + // + // `current_chunks` is the union of the raw file index, suspect entries included, and the + // legacy-only set, which is exactly the question. An error reads as not-empty on purpose: + // being wrong that way costs a `retire_current` where `clear_all` would have done, which + // stops advertising the root and stays answerable until the gossip TTL lapses. Being wrong + // the other way costs a trust penalty on a node that did nothing. + let storage_empty = stored_keys.is_empty() && storage.current_chunks().unwrap_or(1) == 0; let self_id = *p2p.peer_id(); let mut keys = Vec::with_capacity(stored_keys.len()); for k in stored_keys { diff --git a/src/storage/chunk_store.rs b/src/storage/chunk_store.rs index fd4c7435..14f15e11 100644 --- a/src/storage/chunk_store.rs +++ b/src/storage/chunk_store.rs @@ -1119,31 +1119,6 @@ impl ChunkStore { } Err(e) => { let message = format!("{e}"); - // Bigger than this build will ever serve. The legacy store took it - // through an API with no size bound; the file store will not, and no - // amount of retrying changes that. Counted as unusable and removed, - // like a record whose bytes do not match, or one such record would - // stop this node and every node sharing its disk from ever reclaiming - // space. - if message.contains("byte maximum") { - warn!( - "Chunk {} in the legacy environment is larger than this build \ - will store; removing it. It cannot be served either way.", - hex::encode(key) - ); - match legacy.lmdb.delete(key).await { - Ok(_) => { - legacy.only.write().remove(key); - report.unusable += 1; - } - Err(e) => warn!( - "Oversized chunk {} could not be removed from the legacy \ - environment: {e}. The environment stays.", - hex::encode(key) - ), - } - continue; - } if message.contains("Content address mismatch") { // The legacy bytes do not hash to their own key, so this chunk // cannot be reproduced and was never servable. Stop advertising @@ -4724,6 +4699,63 @@ mod tests { assert!(store.has_legacy()); } + /// A legacy value too large for the file store is refused, never deleted. + /// + /// It cannot happen. `MAX_CHUNK_SIZE` has been 4 MiB since the protocol's first commit; + /// the chunk store and its size check arrived together in v0.4.0; replication arrived + /// already carrying the same check on its receive and fetch paths; and every production + /// caller of `put` is guarded. `FileStore::put` also checks the address before the size, + /// so reaching it needs a value over the ceiling that hashes to its own key. + /// + /// The migration used to answer that impossible case by deleting the record, with no + /// possession check and nothing able to replace it. The branch is gone. What is left is + /// the generic error path, which refuses the value, names the key and the size, releases + /// the volume lock so no other node on the disk is held up, and retries. A node that + /// stops making progress and says why is a page; a chunk deleted on a warning is nothing. + #[tokio::test] + async fn an_oversized_legacy_value_is_refused_rather_than_destroyed() { + let dir = TempDir::new().expect("temp dir"); + let content = vec![0xA5u8; crate::ant_protocol::MAX_CHUNK_SIZE + 4096]; + let addr = crate::client::compute_address(&content); + { + let lmdb = LmdbStorage::new(LmdbStorageConfig { + root_dir: dir.path().to_path_buf(), + verify_on_read: false, + // Bounded rather than derived. A derived map is sized from free disk, and + // this is the one legacy fixture in the suite that writes a whole oversized + // chunk into it; the unit tests run in the same job as the test that measures + // how much disk retirement gives back, and an env sized from the volume is + // enough to move that measurement. + max_map_size: 64 * 1024 * 1024, + disk_reserve: 0, + }) + .await + .expect("open legacy"); + lmdb.put(&addr, &content).await.expect("put"); + lmdb.wait_idle().await; + } + let store = open(&dir).await; + + let err = store + .copy_batch(&store.legacy_only_keys(), 0, 0, &never_cancelled()) + .await + .expect_err("an oversized value must fail the pass, not be swallowed"); + assert!( + format!("{err}").contains("byte maximum"), + "the error must say why it refused, got: {err}" + ); + assert_eq!( + store.legacy_only_keys(), + vec![addr], + "the key must stay on the copier's list" + ); + assert_eq!( + store.get(&addr).await.expect("get").expect("still stored"), + content, + "and the value must still be there" + ); + } + /// A legacy record whose bytes do not hash to its key is removed, not passed around. /// /// Leaving it in the environment while dropping it from the key set puts it in @@ -4779,24 +4811,32 @@ mod tests { } #[test] - fn the_release_switches_are_never_written_to_an_operator_config_file() { - // A node writes its effective configuration back to disk. If these round-tripped, - // R1's values would be baked into every operator's file and the next release - // would change nothing. + fn no_migration_setting_is_ever_written_to_an_operator_config_file() { + // A node writes its effective configuration back to disk. When these round-tripped, + // one release's values were baked into every operator's file and the next release + // changed nothing. This used to hold for the release switches alone, with the + // schedule left as genuine operator controls; the whole struct is a build constant + // now, because the release that deletes the old store can only assume every node ran + // the same schedule while no file can change one. let mut config = MigrationConfig::default(); config.retire_legacy = !config.retire_legacy; config.allow_shed = false; config.shed_hold_hours = 5; + config.wave_hours = 9999; let encoded = toml::to_string(&config).expect("encode"); - assert!(!encoded.contains("retire_legacy"), "{encoded}"); + assert_eq!( + encoded.trim(), + "", + "a migration section must not be written out at all, got: {encoded}" + ); let decoded: MigrationConfig = toml::from_str(&encoded).expect("decode"); - let fresh = MigrationConfig::default(); - assert_eq!(decoded.retire_legacy, fresh.retire_legacy); - // Genuine operator controls do survive. - assert!(!decoded.allow_shed); - assert_eq!(decoded.shed_hold_hours, 5); + assert_eq!( + decoded, + MigrationConfig::default(), + "every field must come back from the build, not from the file" + ); } #[test] diff --git a/src/storage/migration.rs b/src/storage/migration.rs index 3087941d..734512a1 100644 --- a/src/storage/migration.rs +++ b/src/storage/migration.rs @@ -98,18 +98,36 @@ const VOLUME_LOCK_COOLDOWN: Duration = Duration::from_secs(120); /// rotation, which is what makes the retention window meaningful. pub const REQUIRED_REBUILDS_BEFORE_RETIRE: u32 = 2; -/// Operator-facing controls for the migration. -// Four independent switches, three of which are operator controls and one of which is a -// release constant. Collapsing them into an enum would tie choices together that are -// deliberately separate. +/// The migration's policy and timing, as release constants. +/// +/// **None of this is read from a node's configuration file.** Every field is `serde(skip)`, +/// so a value in `config.toml` is ignored and a node cannot be configured out of migrating +/// or into a schedule of its own. The release that deletes the old store has to be able to +/// assume every node ran the same schedule, and it cannot assume that while the schedule +/// belongs to whoever last edited a file. +/// +/// What this buys is one schedule, not a guarantee that it completes. Several waits in the +/// migration still have no deadline — a close group that never acknowledges a commitment, a +/// rebuild that keeps failing, a wedged read, a foreign holder of the volume lock — and a +/// node in one of those keeps both stores indefinitely. That is what the fleet signal is for: +/// those nodes are meant to be counted, not assumed away. +/// +/// A node's effective configuration is written back to disk, so these were also the fields +/// most likely to be baked into an operator's file by one release and then honoured by the +/// next. Skipping them on the way in and on the way out ends both halves of that. +/// +/// The fields stay public because tests set them directly to compress a schedule that would +/// otherwise take days. Only the file is refused, not assignment. +// The booleans are deliberately separate rather than an enum: they answer different +// questions and collapsing them would tie choices together that are not tied. #[allow(clippy::struct_excessive_bools)] -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct MigrationConfig { /// Run the background copier at all. /// - /// Turning this off leaves a node reading the union of both stores forever. It never - /// frees the LMDB's disk, so it is an escape hatch rather than a supported mode. - #[serde(default = "default_true")] + /// Release policy, not a setting: see the note on [`MigrationConfig`]. Off, a node reads + /// the union of both stores forever and never frees the LMDB's disk. + #[serde(skip, default = "default_true")] pub enabled: bool, /// Write every new chunk to the legacy environment as well as the file store. @@ -118,14 +136,15 @@ pub struct MigrationConfig { /// the ability to roll the fleet back: a chunk uploaded during the bridge to holders /// that all revert to a pre-migration build would otherwise be gone from every one /// of them. Automatically irrelevant once the legacy environment is retired. - #[serde(default = "default_true")] + #[serde(skip, default = "default_true")] pub dual_write_legacy: bool, /// Allow a node that cannot fit its payload to drop its furthest keys. /// - /// An operator who would rather add disk than shed can set this to `false`. The node - /// then keeps both stores and never frees the LMDB's space. - #[serde(default = "default_true")] + /// Release policy, not a setting: see the note on [`MigrationConfig`]. Off, a node that + /// cannot fit its payload keeps both stores and never frees the LMDB's space, which is + /// why it ships on. + #[serde(skip, default = "default_true")] pub allow_shed: bool, /// Delete `chunks.mdb` once the retirement gate is satisfied. @@ -149,21 +168,21 @@ pub struct MigrationConfig { /// /// Long enough for peers still on a pre-R1 build to upgrade, because one of those /// still penalises a shedder at the full audit weight. - #[serde(default = "default_shed_hold_hours")] + #[serde(skip, default = "default_shed_hold_hours")] pub shed_hold_hours: u64, /// Hours between committing to the file-backed key set and deleting `chunks.mdb`. /// /// Clamped up to [`MIN_RETIRE_DELAY_HOURS`]. Longer buys a rollback window on nodes /// that can afford to hold both copies. - #[serde(default = "default_retire_delay_hours")] + #[serde(skip, default = "default_retire_delay_hours")] pub retire_delay_hours: u64, /// Free megabytes the copier leaves untouched, on top of the disk reserve. /// /// The copier stops here rather than filling to the brink, so a node that is /// mid-migration still has room to accept a chunk it is paid for. - #[serde(default = "default_copier_slack_mb")] + #[serde(skip, default = "default_copier_slack_mb")] pub copier_slack_mb: u64, /// Copy rate ceiling, in mebibytes per second. @@ -171,7 +190,7 @@ pub struct MigrationConfig { /// The quiet responsible audit lane is where audit timeouts actually cost trust, and /// an unthrottled copier competing with it for I/O is the fastest way to turn a /// storage migration into an audit incident. - #[serde(default = "default_copier_throttle_mib_per_sec")] + #[serde(skip, default = "default_copier_throttle_mib_per_sec")] pub copier_throttle_mib_per_sec: u64, /// Hours between one migration wave opening and the next. @@ -181,15 +200,15 @@ pub struct MigrationConfig { /// long a wave gets to finish copying, retiring and refetching before the next one may /// start. Only nodes that have to give something up wait for their wave; a node with /// room migrates immediately. - #[serde(default = "default_wave_hours")] + #[serde(skip, default = "default_wave_hours")] pub wave_hours: u64, /// Seconds between copier ticks. - #[serde(default = "default_tick_secs")] + #[serde(skip, default = "default_tick_secs")] pub tick_secs: u64, /// Chunks copied per tick before yielding. - #[serde(default = "default_batch_chunks")] + #[serde(skip, default = "default_batch_chunks")] pub batch_chunks: usize, /// Where the volume lock lives, overriding the filesystem this node's root sits on. @@ -263,7 +282,7 @@ const fn default_retire_delay_hours() -> u64 { } const fn default_wave_hours() -> u64 { - 24 + 12 } const fn default_copier_slack_mb() -> u64 { @@ -769,8 +788,7 @@ pub struct CopyReport { pub copied: u64, /// Bytes copied. pub bytes: u64, - /// Keys skipped because the legacy bytes did not hash to their address, or were - /// larger than a chunk may be. + /// Keys dropped because the legacy bytes did not hash to their own address. pub unusable: u64, /// Keys that could not be copied for a reason that may clear on a later pass. pub failed: u64, @@ -841,8 +859,8 @@ pub fn wave_has_opened(state: &MigrationState, config: &MigrationConfig, wave: u /// When a given wave opens, in Unix seconds. /// /// Measured from the END of the shed hold, not from first start. Measured from the start -/// the two settings cancel each other out: with a 72 hour hold and 24 hour waves, waves -/// would open at 0, 24, 48 and 72 hours while nothing at all may shed until hour 72, so +/// the two settings cancel each other out: with a 72 hour hold and 12 hour waves, waves +/// would open at 0, 12, 24 and 36 hours while nothing at all may shed until hour 72, so /// every wave would be open the moment the first one could act and the whole close group /// would migrate together. That is the pile-up the waves exist to prevent. #[must_use] @@ -1747,9 +1765,9 @@ async fn evaluate_shed( if !config.allow_shed { warn!( - "This node cannot fit {short_by} chunk(s) in the file store and shedding is \ - turned off. Add disk, or set storage.migration.allow_shed. Until then it \ - keeps serving from both stores and the legacy environment stays." + "This node cannot fit {short_by} chunk(s) in the file store and this build does \ + not shed. Add disk. Until then it keeps serving from both stores and the \ + legacy environment stays." ); return false; } @@ -2985,16 +3003,101 @@ mod tests { ); } + /// A node that was paused comes back with its hold already spent, and that is known. + /// + /// The marker is stamped when the store opens, not when the copier starts, so a node run + /// with `enabled = false` still recorded a first start. This release ignores that setting, + /// so such a node measures its 72-hour hold and all its waves from a stamp that may be + /// weeks old and arrives at the shed path with no hold and no stagger. + /// + /// Pinned rather than fixed: the only way to tell that node from one whose disk filled on + /// day one is that neither has copied anything, and restamping both would delay exactly + /// the nodes with the least room. What is lost is the stagger, not the safety — the + /// possession gate is unchanged and a group shedding together fails each other's checks. + /// This test exists so the behaviour is a decision on the record and not a surprise. + #[test] + fn a_paused_node_resumes_with_its_hold_already_spent() { + let config = MigrationConfig::default(); + assert!( + config.enabled, + "the switch is a build constant now, which is what creates this case" + ); + + let mut state = MigrationState::new(MigrationPhase::Bridging); + // Stamped weeks ago, when the store first opened under a build that then did nothing + // with it. + state.first_start_unix = now_unix().saturating_sub(21 * 24 * 3600); + + for wave in 0..migration_wave_count(7) { + assert!( + wave_has_opened(&state, &config, wave), + "wave {wave} is open on resume, so the stagger is spent as well as the hold" + ); + } + } + + /// Halving the wave spacing under a node that is already migrating never delays it. + /// + /// This is the one thing the schedule change has to promise on upgrade. #216 shipped + /// with 24 hour waves and is already running on a fleet; this release makes them 12. A + /// node picks the new value up mid-migration, against a `first_start_unix` that was + /// persisted under the old one, so the question is whether any node's wave can move + /// later — a node that had been cleared to shed becoming blocked again, after it may + /// already have acted on the earlier answer. + /// + /// It cannot: the two schedules share the same base and the same wave index, and only + /// the multiplier shrinks, so every wave opens at or before where it did. What the + /// change does do is compress the gaps, and while the fleet is mixed two nodes on + /// different builds can land in the same slot (wave 2 at 12 hours opens exactly when + /// wave 1 at 24 does). That is churn, not loss: the possession gate is what stops a + /// chunk being given up, and it is unchanged. + #[test] + fn halving_the_wave_spacing_never_moves_a_node_backwards() { + let old = MigrationConfig { + wave_hours: 24, + ..MigrationConfig::default() + }; + let new = MigrationConfig::default(); + assert_eq!(new.wave_hours, 12); + assert_eq!(old.shed_hold_hours, new.shed_hold_hours); + + let mut state = MigrationState::new(MigrationPhase::Bridging); + // Across every wave a close group of seven can produce, and every age from before + // the hold to well past the last wave. + for wave in 0..migration_wave_count(7) { + for hours_ago in 0..(72 + 24 * 8) { + state.first_start_unix = now_unix().saturating_sub(hours_ago * 3600); + let then = wave_opens_at(&state, &old, wave); + let now = wave_opens_at(&state, &new, wave); + assert!( + now <= then, + "wave {wave} opens later under the new schedule at {hours_ago}h old: \ + {now} > {then}" + ); + if wave_has_opened(&state, &old, wave) { + assert!( + wave_has_opened(&state, &new, wave), + "wave {wave} had opened at {hours_ago}h old and the new schedule \ + closed it again" + ); + } + } + } + } + #[test] fn waves_are_actually_staggered_under_the_shipped_defaults() { // The combination is what matters, not either setting alone. Measured from first - // start, a 72 hour hold and 24 hour waves cancel out: waves would open at 0, 24, - // 48 and 72 hours while nothing may shed until 72, so every wave is open the + // start, a 72 hour hold and 12 hour waves cancel out: waves would open at 0, 12, + // 24 and 36 hours while nothing may shed until 72, so every wave is open the // moment the first one can act and the whole close group moves together. Measured // from the end of the hold, they stagger as intended. let config = MigrationConfig::default(); assert_eq!(config.shed_hold_hours, 72); - assert_eq!(config.wave_hours, 24); + // Pinned, because the whole schedule has to fit inside the two weeks between this + // release and the one that deletes the old store, and the shipped value is the only + // one that matters now the field is no longer read from anyone's file. + assert_eq!(config.wave_hours, 12); let mut state = MigrationState::new(MigrationPhase::Bridging); let waves = migration_wave_count(7); @@ -3021,7 +3124,8 @@ mod tests { // Each later wave opens one wave_hours after the one before it. for open in 1..waves { - state.first_start_unix = now_unix().saturating_sub((72 + open * 24) * 3600 + 60); + state.first_start_unix = + now_unix().saturating_sub((72 + open * config.wave_hours) * 3600 + 60); for w in 0..=open { assert!(wave_has_opened(&state, &config, w)); } diff --git a/src/storage/migration_signal.rs b/src/storage/migration_signal.rs new file mode 100644 index 00000000..c9594e5f --- /dev/null +++ b/src/storage/migration_signal.rs @@ -0,0 +1,699 @@ +//! What a node tells the network about its move off the old chunk store. +//! +//! The release that finally deletes the old store has to be published at a moment when the +//! fleet has finished moving, and "the fleet has finished" is not something a calendar can +//! establish. Nor can our own logs: they cover the nodes we run, and the nodes most likely +//! to still be carrying a `chunks.mdb` are the ones we do not. +//! +//! So a node says so itself, in the one field every peer already sees. `saorsa-core` sends a +//! user agent string with every signed message and keeps each peer's, so any node can ask +//! what its neighbours are. Putting the answer there costs no new message, no new field and +//! no protocol version: it is a different value in a string that was already on the wire. +//! +//! Two rules the string has to obey. It must still begin `node/`, because that prefix is +//! what `saorsa-core` uses to decide whether a peer is a DHT participant at all, and a node +//! that loses it stops being routed to. And the three states must never be folded into two: +//! a directory this node could not read is not the same as one that is not there, and +//! reading "cannot tell" as "finished" is how a gate comes back clean over a fleet that is +//! not. +//! +//! What is deliberately NOT here: anything about whether storage is switched off. A node +//! with `storage.enabled = false` never opens a store, but the old environment is still on +//! its disk and the release that deletes it will still find it. The question this answers is +//! about the filesystem, so it is asked of the filesystem, whatever the node was configured +//! to do with it. + +use std::path::Path; +use std::sync::{Arc, Weak}; +use std::time::Duration; + +use saorsa_core::P2PNode; +use tokio_util::sync::CancellationToken; + +use crate::logging::{info, warn}; + +/// How often a node says where it is and what it can see. +/// +/// Often enough that a reading is never many hours stale, rarely enough that it is a line +/// an operator can read rather than a stream. It is a heartbeat as much as a count: a node +/// that stops saying anything is a node the release gate must treat as unfinished, and it +/// can only do that if a healthy node says something on a known cadence. +const REPORT_INTERVAL: Duration = Duration::from_secs(15 * 60); + +/// The directory the old chunk store lives in. +const LEGACY_ENV_DIR: &str = "chunks.mdb"; + +/// What retirement renames it to before deleting it. +const RETIRED_SUFFIX: &str = ".retired"; + +/// The file retirement writes inside a directory to say it has finished with it. +const RETIRED_MARKER: &str = "RETIRED"; + +/// The token that carries the state, so a reader can find it wherever it sits. +const SIGNAL_PREFIX: &str = "migration/"; + +/// Where this node is in the move off the old chunk store, as seen from its own disk. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum MigrationSignal { + /// Something is still there that this node has not finished with. + Legacy, + /// Nothing is, or only the harmless remains of a cleanup that did not quite finish. + Files, + /// The disk could not be read well enough to say. Never folded into either answer. + Unknown, +} + +impl MigrationSignal { + /// The token this state appears as on the wire. + const fn token(self) -> &'static str { + match self { + Self::Legacy => "legacy", + Self::Files => "files", + Self::Unknown => "unknown", + } + } + + /// Read the state off this node's own disk. + /// + /// Cheap enough to call before the transport is built, which is where it has to be + /// called: the user agent is fixed when the transport is constructed. + #[must_use] + pub fn from_disk(root_dir: &Path) -> Self { + let Ok(dirs) = legacy_directories(root_dir) else { + return Self::Unknown; + }; + let mut answer = Self::Files; + for dir in dirs { + match classify(&dir) { + // Finished with, or empty, which is what an interrupted cleanup leaves. + // Neither holds a chunk, so neither makes this node unfinished. + Leftover::Harmless => {} + Leftover::Holding => return Self::Legacy, + // Keep looking: a directory further down the list may still be holding + // chunks, and that is the stronger answer of the two. + Leftover::Unreadable => answer = Self::Unknown, + } + } + answer + } +} + +/// What one leftover directory means for the node carrying it. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum Leftover { + /// It holds chunks this node has not moved. + Holding, + /// It holds nothing, or it carries the mark that says it was finished with. + Harmless, + /// It could not be read well enough to say which. + Unreadable, +} + +/// Every leftover of the old chunk store under `root_dir`, live name and tombstones alike. +/// +/// The names are matched exactly rather than by prefix. Retirement only ever creates +/// `chunks.mdb.retired` or `chunks.mdb.retired.`, and a prefix match would also claim a +/// directory somebody else put there, which matters because a later release deletes what +/// this list returns. +/// +/// An entry that cannot be read is returned rather than skipped, so it becomes `Unknown` +/// rather than silently becoming `Files`. +fn legacy_directories(root_dir: &Path) -> Result, Unreadable> { + let mut found = Vec::new(); + + // `symlink_metadata`, not `try_exists`: the latter follows links, so a dangling or + // looping one at the live name would read as nothing being there. + let live = root_dir.join(LEGACY_ENV_DIR); + // An error is not an absence: a live name that cannot be queried hides an environment + // that may well be there, so it goes on the list and becomes `Unknown` rather than + // quietly becoming `Files`. + if !matches!(std::fs::symlink_metadata(&live), Err(ref e) if e.kind() == std::io::ErrorKind::NotFound) + { + found.push(live); + } + + let entries = match std::fs::read_dir(root_dir) { + Ok(entries) => entries, + // A root that is not there yet holds nothing, which is every node starting for the + // first time. That is an answer. + Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(found), + // One that cannot be listed hides every tombstone in it, so there is no answer to + // give. Saying so is the whole reason `Unknown` exists. + Err(_) => return Err(Unreadable), + }; + for entry in entries { + // Nor is one unreadable entry evidence that there is nothing behind it. An earlier + // version of this pushed a made-up path here so the caller would classify it, and a + // made-up path that happens not to exist classifies as harmless: one unreadable + // directory entry could hide a real tombstone and still produce `files`, which is + // exactly the false green a release gate must not be able to show. + let Ok(entry) = entry else { + return Err(Unreadable); + }; + if entry.file_name().to_str().is_some_and(is_tombstone_name) { + found.push(entry.path()); + } + } + Ok(found) +} + +/// There is no answer to give about this root. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +struct Unreadable; + +/// The most tombstones one root can hold, matching what retirement will ever create. +const MAX_TOMBSTONES: u32 = 64; + +/// Is this a name retirement gives a tombstone? +/// +/// `chunks.mdb.retired`, or that plus `.` for `n` in `1..=64`, written the way retirement +/// writes it. The bounds are not decoration: retirement only ever counts up to 64, so `.65` +/// and `.007` are names it cannot have produced, and this list becomes a list of directories +/// a later release deletes. +fn is_tombstone_name(name: &str) -> bool { + let base = format!("{LEGACY_ENV_DIR}{RETIRED_SUFFIX}"); + if name == base { + return true; + } + let Some(suffix) = name.strip_prefix(&format!("{base}.")) else { + return false; + }; + // Parsed and then written back out, so a leading zero or a plus sign fails to match + // itself: `"007".parse::()` is happily 7, and `chunks.mdb.retired.007` is not a + // name anything here created. + suffix + .parse::() + .is_ok_and(|n| (1..=MAX_TOMBSTONES).contains(&n) && suffix == n.to_string()) +} + +/// What one directory says about itself. +fn classify(dir: &Path) -> Leftover { + match std::fs::symlink_metadata(dir) { + // A link is never treated as finished with, whatever it points at: the mark would + // have been written through it into a directory that is not this node's. It is also + // never followed to see what is behind it. + Ok(meta) if meta.file_type().is_symlink() => return Leftover::Holding, + Ok(_) => {} + Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Leftover::Harmless, + Err(_) => return Leftover::Unreadable, + } + // A regular file, not merely something at that name. Retirement writes the mark with + // `create_new`, so it is always an ordinary file; a directory, a link, a FIFO or anything + // else wearing the name is not evidence of anything, and this answer is what decides + // whether a later release deletes the chunks underneath it. + match std::fs::symlink_metadata(dir.join(RETIRED_MARKER)) { + Ok(meta) if meta.is_file() => return Leftover::Harmless, + Ok(_) => {} + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(_) => return Leftover::Unreadable, + } + // No mark. A directory with nothing in it holds no chunks, so it cannot be hiding any: + // that is what a cleanup interrupted between emptying a tombstone and removing it + // leaves behind. + std::fs::read_dir(dir).map_or(Leftover::Unreadable, |mut entries| { + if entries.next().is_none() { + Leftover::Harmless + } else { + Leftover::Holding + } + }) +} + +/// The user agent this node announces itself with. +/// +/// Keeps the `node/` prefix `saorsa-core` gates DHT membership on, reports this build's +/// version rather than the transport's, because that is the one a release decision is made +/// about, and carries the migration state as its own token. +#[must_use] +pub fn user_agent(signal: MigrationSignal) -> String { + format!( + "node/{} {SIGNAL_PREFIX}{}", + env!("CARGO_PKG_VERSION"), + signal.token() + ) +} + +/// What a peer's user agent says about that peer. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum PeerMigrationState { + /// It says it still has an old chunk store. + Legacy, + /// It says it has finished. + Files, + /// It says it cannot tell. + Unknown, + /// It says nothing, so it is running a build from before this was reported. Counted on + /// its own rather than with the finished ones: silence is not completion. + Unreported, + /// Not a node at all. Clients connect and announce themselves too, and counting them as + /// nodes that never reported would make every reading look worse than it is. + NotANode, +} + +/// Read a peer's user agent. +#[must_use] +pub fn peer_state(user_agent: &str) -> PeerMigrationState { + if !user_agent.starts_with("node/") { + return PeerMigrationState::NotANode; + } + for token in user_agent.split_whitespace() { + let Some(state) = token.strip_prefix(SIGNAL_PREFIX) else { + continue; + }; + return match state { + "legacy" => PeerMigrationState::Legacy, + "files" => PeerMigrationState::Files, + // A token we do not recognise is a build that reports something this one has + // never heard of. That is not "finished". + _ => PeerMigrationState::Unknown, + }; + } + PeerMigrationState::Unreported +} + +/// One tally of what a node can see around it. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct PeerTally { + /// Peers that still have an old chunk store. + pub legacy: usize, + /// Peers that have finished. + pub files: usize, + /// Peers that could not tell, or that answered with something this build does not know. + pub unknown: usize, + /// Peers running a build from before this was reported. + pub unreported: usize, +} + +impl PeerTally { + /// Peers that are not evidence the fleet has finished. + #[must_use] + pub const fn outstanding(self) -> usize { + self.legacy + self.unknown + self.unreported + } + + fn add(&mut self, state: PeerMigrationState) { + match state { + PeerMigrationState::Legacy => self.legacy += 1, + PeerMigrationState::Files => self.files += 1, + PeerMigrationState::Unknown => self.unknown += 1, + PeerMigrationState::Unreported => self.unreported += 1, + // Deliberately not counted at all. A client is not a node that failed to + // report, and putting it in any of the buckets above would make every reading + // worse than it is. + PeerMigrationState::NotANode => {} + } + } +} + +/// Count what this node can see of its neighbours. +/// +/// These are edges, not nodes: two of our nodes connected to the same peer both report it, +/// and a peer nobody is connected to is in nobody's count. That is why each line carries the +/// observer, so whoever adds them up can decide what a peer is worth rather than trusting an +/// arithmetic sum. +pub async fn tally_peers(p2p: &Arc) -> PeerTally { + let mut tally = PeerTally::default(); + let transport = p2p.transport(); + let observer = p2p.peer_id().to_hex(); + for peer in transport.connected_peers().await { + // No agent recorded is not the same as a peer that reported nothing, but it is + // just as far from evidence of completion, so it lands in the same bucket rather + // than being skipped. + let agent = transport.peer_user_agent(&peer).await; + let state = agent + .as_deref() + .map_or(PeerMigrationState::Unreported, peer_state); + // One line per peer, not just the totals. What a node sees are edges: two of our + // nodes connected to the same peer both report it, and a peer nobody is connected to + // is in nobody's count. Summing the totals across the fleet therefore counts some + // nodes twice and others never, which is not a number a release decision can rest + // on. With the observer, the peer and the moment on each line, whoever adds them up + // can deduplicate by peer and apply their own freshness rule; without them, they + // cannot. + // + // At `info`, not `debug`. Nodes run at `info` (`cli.rs:95`), so the same line at + // `debug` is written nowhere the gate can read it, and the release would be decided + // on the aggregates alone — which is the number that cannot be deduplicated. A line + // nobody emits is not a signal. + // + // Only for peers that are NOT finished. The gate asks which distinct nodes are still + // outstanding, so those are the ones that need naming; a fleet that has finished + // emits none of these at all, and the cost falls away as the answer arrives rather + // than peaking when it does. The aggregate line below is emitted either way and + // carries the finished count, so the denominator does not go missing with them, and + // a node that has gone quiet is still distinguishable from a node with nothing to + // report. + if state != PeerMigrationState::Files && state != PeerMigrationState::NotANode { + info!( + migration_event = "peer_state", + observer = %observer, + peer = %peer.to_hex(), + state = peer_state_token(state), + agent = agent.as_deref().unwrap_or("none"), + "Storage migration: peer {} is {}", + peer.to_hex(), + peer_state_token(state) + ); + } + tally.add(state); + } + tally +} + +/// The token a peer's state is reported as, so the aggregate line and the per-peer lines +/// cannot drift apart. +/// +/// Only ever read by a log line, so it goes when the logging feature does. +#[cfg_attr(not(feature = "logging"), allow(dead_code))] +const fn peer_state_token(state: PeerMigrationState) -> &'static str { + match state { + PeerMigrationState::Legacy => "legacy", + PeerMigrationState::Files => "files", + PeerMigrationState::Unknown => "unknown", + PeerMigrationState::Unreported => "unreported", + PeerMigrationState::NotANode => "not-a-node", + } +} + +/// Say where this node is, and what it can see, until it shuts down. +/// +/// Structured rather than prose, because the thing that reads it is a query and not a +/// person: a release decision is "no node says `legacy`, and every node we expect to hear +/// from said something recently", and neither half of that can be answered from a sentence. +pub async fn report_until_shutdown( + p2p: Weak, + root_dir: std::path::PathBuf, + shutdown: CancellationToken, +) { + loop { + // Wait first, report second. A node that has just started has no peers to describe + // and nothing has changed on its disk since the user agent was built from it, so the + // first pass would be pure startup cost on the busiest moment in a node's life. + tokio::select! { + () = shutdown.cancelled() => return, + () = tokio::time::sleep(REPORT_INTERVAL) => {} + } + // A **weak** handle, upgraded per pass and dropped again before the sleep. This task + // observes a node; it must never be the reason one stays alive. Holding a strong one + // keeps the node, and with it its bound transport, for as long as this task runs, so + // any path that drops a node without cancelling its token would leak a live port + // rather than a stopped node. Nothing is left to notice that but the next thing that + // fails to bind. + let Some(p2p) = p2p.upgrade() else { + return; + }; + // Re-read the disk every time rather than reusing what the user agent was built + // from. The agent is fixed when the transport is built; this is not, so a node that + // finishes migrating says so here within the interval instead of at its next + // restart. + let own = MigrationSignal::from_disk(&root_dir); + let peers = tally_peers(&p2p).await; + if own == MigrationSignal::Legacy || own == MigrationSignal::Unknown { + warn!( + migration_event = "signal", + state = own.token(), + peers_legacy = peers.legacy, + peers_unknown = peers.unknown, + peers_unreported = peers.unreported, + peers_files = peers.files, + "This node still has an old chunk store ({}). {} of the {} node(s) it can \ + see are not finished either.", + own.token(), + peers.outstanding(), + peers.outstanding() + peers.files + ); + } else { + info!( + migration_event = "signal", + state = own.token(), + peers_legacy = peers.legacy, + peers_unknown = peers.unknown, + peers_unreported = peers.unreported, + peers_files = peers.files, + "Storage migration: this node is done; {} of the {} node(s) it can see are \ + not.", + peers.outstanding(), + peers.outstanding() + peers.files + ); + } + // Dropped before the next wait, so the node is not held across the interval. + drop(p2p); + } +} + +#[cfg(test)] +#[allow(clippy::unwrap_used, clippy::expect_used)] +mod tests { + + /// The reporter must not keep the node it reports on alive. + /// + /// It holds a weak handle and upgrades per pass. A strong one would keep the node, and + /// with it the transport and the bound port, for as long as the task ran, so any path + /// that dropped a node without cancelling its token would leak a live port instead of + /// stopping a node. Nothing notices that until the next bind fails, somewhere else, + /// much later. + #[tokio::test] + async fn the_reporter_lets_go_of_a_node_that_was_dropped() { + let dir = tempfile::tempdir().expect("temp dir"); + let root = dir.path().to_path_buf(); + let shutdown = CancellationToken::new(); + + // Stand in for the node: what matters is that the task holds no strong reference, + // so the count of strong holders does not rise when the reporter starts, and the + // reporter stops on its own once the last real holder goes. + let owner = Arc::new(()); + let weak = Arc::downgrade(&owner); + assert_eq!(Arc::strong_count(&owner), 1); + + let handle = tokio::spawn({ + let weak = weak.clone(); + let shutdown = shutdown.clone(); + async move { + loop { + tokio::select! { + () = shutdown.cancelled() => return "cancelled", + () = tokio::time::sleep(std::time::Duration::from_millis(5)) => {} + } + let Some(up) = weak.upgrade() else { + return "node went away"; + }; + drop(up); + } + } + }); + + assert_eq!( + Arc::strong_count(&owner), + 1, + "starting the reporter must not add a strong holder" + ); + drop(owner); + let outcome = tokio::time::timeout(std::time::Duration::from_secs(5), handle) + .await + .expect("the reporter must stop on its own") + .expect("task must not panic"); + assert_eq!( + outcome, "node went away", + "the reporter must stop when the node is gone, not wait for a cancellation \ + nobody sends" + ); + let _ = root; + } + + use super::*; + use tempfile::TempDir; + + fn dir_with(root: &Path, name: &str) -> std::path::PathBuf { + let path = root.join(name); + std::fs::create_dir_all(&path).unwrap(); + path + } + + #[test] + fn a_node_with_nothing_on_disk_has_finished() { + let root = TempDir::new().unwrap(); + assert_eq!( + MigrationSignal::from_disk(root.path()), + MigrationSignal::Files + ); + } + + #[test] + fn a_root_that_does_not_exist_yet_has_finished() { + let root = TempDir::new().unwrap(); + let never = root.path().join("not-created"); + assert_eq!(MigrationSignal::from_disk(&never), MigrationSignal::Files); + } + + #[test] + fn a_live_environment_with_chunks_in_it_has_not() { + let root = TempDir::new().unwrap(); + let env = dir_with(root.path(), LEGACY_ENV_DIR); + std::fs::write(env.join("data.mdb"), b"chunks").unwrap(); + assert_eq!( + MigrationSignal::from_disk(root.path()), + MigrationSignal::Legacy + ); + } + + #[test] + fn a_marked_leftover_has_finished() { + let root = TempDir::new().unwrap(); + let env = dir_with(root.path(), LEGACY_ENV_DIR); + std::fs::write(env.join("data.mdb"), b"chunks").unwrap(); + std::fs::write(env.join(RETIRED_MARKER), b"").unwrap(); + assert_eq!( + MigrationSignal::from_disk(root.path()), + MigrationSignal::Files + ); + } + + #[test] + fn an_empty_leftover_has_finished() { + // What a cleanup interrupted between emptying a tombstone and removing it leaves. + let root = TempDir::new().unwrap(); + dir_with(root.path(), "chunks.mdb.retired"); + assert_eq!( + MigrationSignal::from_disk(root.path()), + MigrationSignal::Files + ); + } + + #[test] + fn a_tombstone_with_chunks_in_it_has_not() { + // A crash between the rename and the mark leaves an intact environment wearing a + // retired-looking name. What it is called is not evidence. + let root = TempDir::new().unwrap(); + let tomb = dir_with(root.path(), "chunks.mdb.retired.3"); + std::fs::write(tomb.join("data.mdb"), b"chunks").unwrap(); + assert_eq!( + MigrationSignal::from_disk(root.path()), + MigrationSignal::Legacy + ); + } + + #[test] + fn an_entry_that_cannot_be_read_is_never_read_as_finished() { + // An earlier version pushed a made-up path when an entry could not be read, so the + // caller would classify it. A made-up path that happens not to exist classifies as + // harmless, so one unreadable entry could hide a real tombstone and still answer + // `files`. A gate that can come back green over a fleet that has not finished is + // worse than no gate. + let root = TempDir::new().unwrap(); + let unreadable = root.path().join("locked"); + std::fs::create_dir_all(&unreadable).unwrap(); + let tomb = unreadable.join("chunks.mdb.retired"); + std::fs::create_dir_all(&tomb).unwrap(); + std::fs::write(tomb.join("data.mdb"), b"chunks").unwrap(); + + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&unreadable, std::fs::Permissions::from_mode(0o000)).unwrap(); + let answer = MigrationSignal::from_disk(&unreadable); + std::fs::set_permissions(&unreadable, std::fs::Permissions::from_mode(0o755)).unwrap(); + assert_eq!( + answer, + MigrationSignal::Unknown, + "a root that cannot be listed must never answer that it has finished" + ); + } + } + + #[test] + fn a_directory_that_only_looks_like_a_tombstone_is_not_one() { + // This list is eventually a list of directories a later release deletes, so it + // matches the names retirement actually creates and nothing else. + assert!(is_tombstone_name("chunks.mdb.retired")); + assert!(is_tombstone_name("chunks.mdb.retired.1")); + assert!(is_tombstone_name("chunks.mdb.retired.42")); + assert!(!is_tombstone_name("chunks.mdb.retired-mine")); + assert!(!is_tombstone_name("chunks.mdb.retired.")); + assert!(!is_tombstone_name("chunks.mdb.retired.backup")); + assert!(!is_tombstone_name("chunks.mdb")); + // Names retirement counts up to, and names it never reaches. `.007` parses as 7 and + // is still not a name anything wrote. + assert!(is_tombstone_name("chunks.mdb.retired.64")); + assert!(!is_tombstone_name("chunks.mdb.retired.65")); + assert!(!is_tombstone_name("chunks.mdb.retired.0")); + assert!(!is_tombstone_name("chunks.mdb.retired.007")); + assert!(!is_tombstone_name("chunks.mdb.retired.+1")); + assert!(!is_tombstone_name("chunks.mdb.retired.999999")); + + let root = TempDir::new().unwrap(); + let mine = dir_with(root.path(), "chunks.mdb.retired-mine"); + std::fs::write(mine.join("data.mdb"), b"somebody else's").unwrap(); + assert_eq!( + MigrationSignal::from_disk(root.path()), + MigrationSignal::Files + ); + } + + #[test] + fn a_linked_environment_is_never_read_as_finished() { + // The mark would have been written through the link into a directory this node + // does not own, so a link is never evidence that anything was finished with, and + // what it points at is never followed. + let root = TempDir::new().unwrap(); + let elsewhere = dir_with(root.path(), "elsewhere"); + std::fs::write(elsewhere.join(RETIRED_MARKER), b"").unwrap(); + #[cfg(unix)] + { + std::os::unix::fs::symlink(&elsewhere, root.path().join(LEGACY_ENV_DIR)).unwrap(); + assert_eq!( + MigrationSignal::from_disk(root.path()), + MigrationSignal::Legacy + ); + } + } + + #[test] + fn the_user_agent_keeps_the_prefix_that_gates_dht_membership() { + // saorsa-core decides whether a peer is a DHT participant by this prefix alone. A + // node that loses it stops being routed to, which is a much worse outcome than not + // reporting at all, so it is worth pinning. + for signal in [ + MigrationSignal::Legacy, + MigrationSignal::Files, + MigrationSignal::Unknown, + ] { + assert!(user_agent(signal).starts_with("node/")); + } + } + + #[test] + fn a_peer_reads_back_what_a_node_announced() { + assert_eq!( + peer_state(&user_agent(MigrationSignal::Legacy)), + PeerMigrationState::Legacy + ); + assert_eq!( + peer_state(&user_agent(MigrationSignal::Files)), + PeerMigrationState::Files + ); + assert_eq!( + peer_state(&user_agent(MigrationSignal::Unknown)), + PeerMigrationState::Unknown + ); + } + + #[test] + fn silence_is_counted_as_silence_and_not_as_completion() { + // The build before this one announces the transport's own agent, with no token of + // ours. Reading that as "finished" is exactly how a gate comes back clean over a + // fleet that has not finished. + assert_eq!(peer_state("node/0.27.0"), PeerMigrationState::Unreported); + assert_eq!( + peer_state("node/0.17.2 migration/something-new"), + PeerMigrationState::Unknown + ); + } + + #[test] + fn a_client_is_not_a_node_that_failed_to_report() { + // Clients authenticate and announce themselves too. Counting them among the peers + // that never reported would make every reading look worse than it is, and the + // count is what a release decision is made on. + assert_eq!(peer_state("client/0.27.0"), PeerMigrationState::NotANode); + } +} diff --git a/src/storage/mod.rs b/src/storage/mod.rs index bda34ac8..4745b283 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -52,6 +52,7 @@ pub(crate) mod file_store; mod handler; pub(crate) mod lmdb; pub mod migration; +pub mod migration_signal; pub use crate::ant_protocol::XorName; pub use chunk_store::{ChunkStore, ChunkStoreConfig, VerifyReport, LEGACY_ENV_DIR}; @@ -61,6 +62,7 @@ pub(crate) use handler::ChunkRequestContext; pub(crate) use lmdb::CapacityVerdict; pub use lmdb::{LmdbStorage, LmdbStorageConfig}; pub use migration::{MigrationConfig, MigrationPhase, MigrationState}; +pub use migration_signal::{peer_state, MigrationSignal, PeerMigrationState}; /// Bytes in one MiB. pub const MIB: u64 = 1024 * 1024; diff --git a/src/upgrade/mod.rs b/src/upgrade/mod.rs index 5502ac43..40efc24f 100644 --- a/src/upgrade/mod.rs +++ b/src/upgrade/mod.rs @@ -13,6 +13,7 @@ mod cache_dir; mod monitor; mod release_cache; mod rollout; +mod rollout_state; mod signature; pub use apply::{AutoApplyUpgrader, RESTART_EXIT_CODE}; @@ -21,6 +22,7 @@ pub use cache_dir::upgrade_cache_dir; pub use monitor::{find_platform_asset, version_from_tag, Asset, GitHubRelease, UpgradeMonitor}; pub use release_cache::ReleaseCache; pub use rollout::StagedRollout; +pub use rollout_state::RolloutState; pub use signature::{ verify_binary_signature, verify_binary_signature_with_key, verify_from_file, verify_from_file_with_key, PUBLIC_KEY_SIZE, SIGNATURE_SIZE, SIGNING_CONTEXT, diff --git a/src/upgrade/monitor.rs b/src/upgrade/monitor.rs index b6d6b0a3..02234d23 100644 --- a/src/upgrade/monitor.rs +++ b/src/upgrade/monitor.rs @@ -12,6 +12,7 @@ use crate::error::{Error, Result}; use crate::logging::{debug, info, warn}; use crate::upgrade::release_cache::ReleaseCache; use crate::upgrade::rollout::StagedRollout; +use crate::upgrade::rollout_state::RolloutState; use crate::upgrade::UpgradeInfo; use semver::Version; use serde::Deserialize; @@ -57,8 +58,15 @@ pub struct UpgradeMonitor { staged_rollout: Option, /// Disk cache for GitHub release metadata (shared across instances). release_cache: Option, - /// When the current pending upgrade was first detected. + /// When the current pending upgrade was first detected, for this process only. pending_upgrade_detected: Option, + /// Where that moment is written down, so it survives a restart. + /// + /// Without it the window restarts every time the node does, and a node that restarts + /// more often than its own delay never reaches the end of one. `None` keeps the old + /// in-process behaviour, which is what tests and any caller without a root directory + /// get. + rollout_state: Option, /// The version of the pending upgrade (for tracking rollout state). pending_upgrade_version: Option, } @@ -94,6 +102,7 @@ impl UpgradeMonitor { staged_rollout: None, release_cache: None, pending_upgrade_detected: None, + rollout_state: None, pending_upgrade_version: None, } } @@ -124,6 +133,18 @@ impl UpgradeMonitor { self } + /// Remember when a release was first seen, under this node's root directory. + /// + /// The rollout delay is measured from the moment a node first sees a version. Kept only + /// in memory, that moment is lost on every restart, so a node that restarts inside its + /// own delay starts the window again and can defer an upgrade indefinitely. The release + /// that later assumes the fleet has had its window cannot assume it while that is true. + #[must_use] + pub fn with_rollout_state(mut self, root_dir: &std::path::Path) -> Self { + self.rollout_state = Some(RolloutState::new(root_dir)); + self + } + /// Create a monitor with a custom current version (for testing). #[cfg(test)] #[must_use] @@ -151,6 +172,7 @@ impl UpgradeMonitor { staged_rollout: None, release_cache: None, pending_upgrade_detected: None, + rollout_state: None, pending_upgrade_version: None, } } @@ -298,35 +320,40 @@ impl UpgradeMonitor { .as_ref() .map_or(true, |v| *v != info.version); + let delay = rollout.calculate_delay_for_version(&info.version); + if is_new_version { // New version detected - start rollout timer self.pending_upgrade_detected = Some(Instant::now()); self.pending_upgrade_version = Some(info.version.clone()); - - let delay = rollout.calculate_delay_for_version(&info.version); - let restart_time = chrono::Utc::now() - + chrono::Duration::from_std(delay).unwrap_or_else(|_| chrono::Duration::hours(1)); info!( new_version = %info.version, delay_hours = delay.as_secs() / 3600, delay_minutes = (delay.as_secs() % 3600) / 60, "New version detected, staged rollout delay calculated" ); - info!( - "Node will stop/restart for upgrade at {}", - restart_time.to_rfc3339() - ); } - // Calculate if we're past the rollout delay - let Some(detected_at) = self.pending_upgrade_detected else { + // How long this node has been waiting. Taken from the record on disk when there is + // one, so a restart does not start the window again: a node restarting more often + // than its own delay would otherwise never reach the end of one, and would sit on an + // old release for as long as it kept restarting. + // + // Every way of failing to establish the answer ends in "upgrade now". A node that + // cannot say when it started waiting cannot show it has finished, and a node left + // behind is a worse outcome than one that upgrades a few hours early. + let Some(elapsed) = self.waited_for(&info.version, rollout.max_delay_hours(), delay) else { // Should not happen, but handle gracefully warn!("Pending upgrade detected but no timestamp recorded"); return Ok(Some(info)); }; - let delay = rollout.calculate_delay_for_version(&info.version); - let elapsed = detected_at.elapsed(); + // Deliberately no "will restart at" line here. The caller already logs the deadline + // from `time_until_upgrade`, which measures the same way this does, and a second one + // said before the download, the signature check and the replacement have succeeded + // promises something this cannot know. The version-detected branch above cannot say + // it either: a restart makes the version look new to this process while the recorded + // moment says most of the wait is already done. if elapsed >= delay { info!( @@ -379,10 +406,12 @@ impl UpgradeMonitor { pub fn time_until_upgrade(&self) -> Option { let rollout = self.staged_rollout.as_ref()?; let version = self.pending_upgrade_version.as_ref()?; - let detected_at = self.pending_upgrade_detected?; - let delay = rollout.calculate_delay_for_version(version); - let elapsed = detected_at.elapsed(); + // The same measurement the readiness check uses. When these were two calculations + // they disagreed after a restart: readiness read the recorded moment and said one + // hour was left, while this read a process-local clock that had just started and + // told the caller to sleep for nearly another day. + let elapsed = self.waited_for(version, rollout.max_delay_hours(), delay)?; if elapsed >= delay { Some(Duration::ZERO) @@ -391,6 +420,36 @@ impl UpgradeMonitor { } } + /// How long this node has been waiting for `version`. + /// + /// One measurement, used by both the readiness check and the sleep the caller takes + /// between checks, because two of them drift apart across a restart and the node ends up + /// eligible and asleep at the same time. + /// + /// From the record on disk when there is one, so a restart does not start the window + /// again. `spent` is what to answer when the record cannot be established at all: a node + /// that cannot say when it began waiting cannot show that it has finished, and being left + /// behind is a far worse outcome than upgrading early with its share of the fleet. + fn waited_for( + &self, + version: &Version, + window_hours: u64, + spent: Duration, + ) -> Option { + if let Some(state) = self.rollout_state.as_ref() { + return Some( + state + .first_seen(version, window_hours) + .map_or(spent, |first_seen| { + Duration::from_secs( + now_unix().unwrap_or(first_seen).saturating_sub(first_seen), + ) + }), + ); + } + self.pending_upgrade_detected.map(|at| at.elapsed()) + } + /// Check if staged rollout is enabled. #[must_use] pub fn has_staged_rollout(&self) -> bool { @@ -681,6 +740,14 @@ fn build_platform_patterns(arch: &str, os: &str) -> Vec { patterns } +/// Now, in Unix seconds. +fn now_unix() -> Option { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .ok() + .map(|d| d.as_secs()) +} + #[cfg(test)] #[allow( clippy::unwrap_used, diff --git a/src/upgrade/rollout_state.rs b/src/upgrade/rollout_state.rs new file mode 100644 index 00000000..6c1c63b8 --- /dev/null +++ b/src/upgrade/rollout_state.rs @@ -0,0 +1,312 @@ +//! When this node first saw the release it is waiting to install, kept across restarts. +//! +//! A staged rollout spreads a release over a window so the fleet does not restart at once. +//! The delay is measured from the moment a node first sees the new version, and that moment +//! used to live only in memory: a node that restarted before its delay ran out started the +//! window again from zero. A node that restarts often enough never reaches the end of it, +//! and quietly stays on an old release for as long as it keeps restarting. That is the +//! difference between "the fleet had two weeks" and "the fleet had two weeks unless it +//! restarted", and only the first of those is something a later release can rely on. +//! +//! So the moment is written down. Deliberately small: one target, one timestamp, no history. +//! +//! Every failure here means "upgrade now" rather than "wait". A node that cannot record when +//! it started waiting has no way to prove it ever finished waiting, and the failure that +//! matters is the one that leaves a node behind, not the one that lets it upgrade a few hours +//! early with its share of the fleet. + +use std::path::{Path, PathBuf}; + +use semver::Version; +use serde::{Deserialize, Serialize}; + +use crate::logging::{debug, warn}; + +/// The file this lives in, under the node's own root. +/// +/// Not the shared upgrade cache: that is keyed by machine and shared by every node on it, +/// and this is one node's place in one window. +const FILE: &str = "upgrade-rollout.json"; + +/// Bumped if the shape changes. An older or newer shape is treated as no record at all, +/// which means the node upgrades rather than waits. +const SCHEMA: u32 = 1; + +/// A clock that has jumped further ahead than this makes the record meaningless. +/// +/// Believing a stamp from the future would have a node wait out a delay that never elapses. +/// Rewriting it costs at most one node upgrading on a fresh window. +const IMPLAUSIBLE_FUTURE_SECS: u64 = 24 * 3600; + +/// What was written down. +#[derive(Debug, Clone, Serialize, Deserialize)] +struct Stamp { + /// Shape of this record. + schema: u32, + /// The release being waited for. + version: String, + /// When this node first saw it, in Unix seconds. + first_seen_unix: u64, + /// The window in force when it was written, so a changed window is visible rather than + /// silently reinterpreting an old stamp. + window_hours: u64, +} + +/// This node's place in the current rollout window. +#[derive(Debug, Clone)] +pub struct RolloutState { + path: PathBuf, +} + +impl RolloutState { + /// Keep the record under this node's root directory. + #[must_use] + pub fn new(root_dir: &Path) -> Self { + Self { + path: root_dir.join(FILE), + } + } + + /// When this node first saw `version`, recording it if this is the first time. + /// + /// `None` means the answer could not be established, and the caller must read that as + /// "the delay has elapsed". Waiting on an answer that cannot be written down is how a + /// node waits forever. + #[must_use] + pub fn first_seen(&self, version: &Version, window_hours: u64) -> Option { + let now = now_unix()?; + match self.read() { + Record::Stamped(stamp) => { + if stamp.version == version.to_string() && stamp.window_hours == window_hours { + if stamp.first_seen_unix <= now.saturating_add(IMPLAUSIBLE_FUTURE_SECS) { + return Some(stamp.first_seen_unix.min(now)); + } + warn!( + "Upgrade rollout: the recorded time for {version} is implausibly far \ + in the future, so it cannot be used to measure a wait. Upgrading \ + without waiting out this node's share of the window." + ); + return None; + } + } + // Nothing written down yet, which is the ordinary first sighting of a release. + Record::Absent => {} + // Something is at that name and it is not a record this build can read. Not + // replaced with a fresh one: writing "now" over it restarts the window, and a + // node whose disk keeps producing unreadable files would restart it on every + // boot and never upgrade at all. Answering "no record" upgrades this node once + // and is done with it. + Record::Unreadable => { + warn!( + "Upgrade rollout: {} cannot be read, so this node cannot tell when it \ + began waiting for {version}. Upgrading without waiting out its share \ + of the window.", + self.path.display() + ); + return None; + } + } + self.write(&Stamp { + schema: SCHEMA, + version: version.to_string(), + first_seen_unix: now, + window_hours, + })?; + Some(now) + } + + /// Read the record. + fn read(&self) -> Record { + let bytes = match std::fs::read(&self.path) { + Ok(bytes) => bytes, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Record::Absent, + Err(_) => return Record::Unreadable, + }; + let Ok(stamp) = serde_json::from_slice::(&bytes) else { + return Record::Unreadable; + }; + if stamp.schema == SCHEMA { + Record::Stamped(stamp) + } else { + debug!( + "Upgrade rollout: {} is schema {}, not {SCHEMA}; treating it as no record", + self.path.display(), + stamp.schema + ); + Record::Unreadable + } + } + + /// Write the record, atomically and durably, or say it could not be written. + /// + /// Flushed rather than just written. The whole point is to survive a restart, and the + /// restart most likely to lose an unflushed file is the abrupt kind this is measuring + /// across. + fn write(&self, stamp: &Stamp) -> Option<()> { + let encoded = serde_json::to_vec(stamp).ok()?; + let temp = self.path.with_extension("json.tmp"); + write_and_flush(&temp, &encoded) + .and_then(|()| std::fs::rename(&temp, &self.path)) + .and_then(|()| flush_dir(self.path.parent().unwrap_or(&self.path))) + .map_err(|e| { + warn!( + "Upgrade rollout: could not record when this node first saw {}: {e}. It \ + will upgrade without waiting out its share of the window rather than \ + wait for a deadline it cannot remember.", + stamp.version + ); + // A failed rename can leave the temporary file behind. Nothing reads it, but + // leaving one per attempt in a node's root is untidy. + let _ = std::fs::remove_file(&temp); + }) + .ok() + } +} + +/// What is at the record's name. +enum Record { + /// A record this build understands. + Stamped(Stamp), + /// Nothing, which is the ordinary first sighting of a release. + Absent, + /// Something that is not a record this build can read. Never quietly replaced: writing + /// a fresh one restarts the window, and a disk that keeps producing unreadable files + /// would restart it on every boot. + Unreadable, +} + +/// Write a file and flush it, so it is there after a power loss and not merely after a +/// clean shutdown. +fn write_and_flush(path: &Path, bytes: &[u8]) -> std::io::Result<()> { + use std::io::Write; + let mut file = std::fs::File::create(path)?; + file.write_all(bytes)?; + file.sync_all() +} + +/// Flush a directory, so the entry naming a file is durable too. +fn flush_dir(dir: &Path) -> std::io::Result<()> { + #[cfg(unix)] + { + std::fs::File::open(dir)?.sync_all() + } + // Off Unix a directory cannot be opened for this, and the rename is the platform's own + // business. Not an error: the alternative is refusing to record anything at all. + #[cfg(not(unix))] + { + let _ = dir; + Ok(()) + } +} + +/// Now, in Unix seconds. +fn now_unix() -> Option { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .ok() + .map(|d| d.as_secs()) +} + +#[cfg(test)] +#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] +mod tests { + use super::*; + use tempfile::TempDir; + + fn version(v: &str) -> Version { + Version::parse(v).unwrap() + } + + #[test] + fn the_window_survives_a_restart() { + // The whole point. Asking twice, as two runs of the same node would, has to give the + // same answer, or a node that restarts often never reaches the end of any window. + let dir = TempDir::new().unwrap(); + let state = RolloutState::new(dir.path()); + let first = state.first_seen(&version("0.17.2"), 24).unwrap(); + + let after_restart = RolloutState::new(dir.path()); + let second = after_restart.first_seen(&version("0.17.2"), 24).unwrap(); + assert_eq!(first, second); + } + + #[test] + fn a_different_release_starts_a_new_window() { + let dir = TempDir::new().unwrap(); + let state = RolloutState::new(dir.path()); + state.first_seen(&version("0.17.2"), 24).unwrap(); + + // Written down as the new target, not silently answered from the old record. + state.first_seen(&version("0.18.0"), 24).unwrap(); + let Record::Stamped(stamp) = state.read() else { + panic!("the new target must have been written down") + }; + assert_eq!(stamp.version, "0.18.0"); + } + + #[test] + fn a_changed_window_starts_a_new_one_too() { + // A stamp says when the wait began; the window says how long it is. Reusing a stamp + // written under a different window silently reinterprets it. + let dir = TempDir::new().unwrap(); + let state = RolloutState::new(dir.path()); + state.first_seen(&version("0.17.2"), 24).unwrap(); + state.first_seen(&version("0.17.2"), 1).unwrap(); + let Record::Stamped(stamp) = state.read() else { + panic!("the new window must have been written down") + }; + assert_eq!(stamp.window_hours, 1); + } + + #[test] + fn a_record_that_cannot_be_written_means_upgrade_rather_than_wait() { + // A node that cannot record when it started waiting cannot prove it ever finished. + // Upgrading early with its share of the fleet is the cheaper failure by far. + let dir = TempDir::new().unwrap(); + let unwritable = dir.path().join("no").join("such").join("directory"); + let state = RolloutState::new(&unwritable); + assert_eq!(state.first_seen(&version("0.17.2"), 24), None); + } + + #[test] + fn a_corrupt_record_means_upgrade_rather_than_a_fresh_window() { + // Not replaced with a stamp of "now". That restarts the window, and a node whose + // disk keeps producing unreadable files would restart it on every boot and never + // upgrade at all. Answering "no record" upgrades this node once and is done. + let dir = TempDir::new().unwrap(); + std::fs::write(dir.path().join(FILE), b"not json").unwrap(); + let state = RolloutState::new(dir.path()); + assert_eq!(state.first_seen(&version("0.17.2"), 24), None); + } + + #[test] + fn a_stamp_from_the_future_does_not_make_the_wait_endless() { + // A clock that jumped forward and back would otherwise leave a node waiting out a + // delay measured from a moment that has not happened yet. + let dir = TempDir::new().unwrap(); + let state = RolloutState::new(dir.path()); + let now = now_unix().unwrap(); + state + .write(&Stamp { + schema: SCHEMA, + version: "0.17.2".into(), + first_seen_unix: now + 10 * IMPLAUSIBLE_FUTURE_SECS, + window_hours: 24, + }) + .unwrap(); + + assert_eq!(state.first_seen(&version("0.17.2"), 24), None); + } + + #[test] + fn a_stamp_from_an_older_shape_means_upgrade_too() { + let dir = TempDir::new().unwrap(); + std::fs::write( + dir.path().join(FILE), + br#"{"schema":0,"version":"0.1.0","first_seen_unix":1,"window_hours":24}"#, + ) + .unwrap(); + let state = RolloutState::new(dir.path()); + assert_eq!(state.first_seen(&version("0.17.2"), 24), None); + } +}