From 0e07a903b9e1393fb001e777d83429d2c219ac79 Mon Sep 17 00:00:00 2001 From: grumbach Date: Wed, 9 Sep 2026 14:24:27 +0900 Subject: [PATCH 1/2] fix(storage): take the migration schedule off config files, and stop a full node clearing its commitments Two changes on top of the release that replaced the LMDB chunk store, both needed before the release that deletes it. Every `MigrationConfig` field is now `serde(skip)`. A `[storage.migration]` section left in an operator's file still parses and is ignored, and nothing is written back out. The release that finally 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: 24 put the worst case at about eight days before copy time, which is itself unbounded. Two consequences, both intended. A node whose file says `enabled = false` migrates anyway, because nobody may opt out of a migration the next release assumes has happened. And 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 that node from one whose disk filled on day one, and restamping both would add three days of waiting to the nodes with the least room. What such a node loses is the stagger, not the safety: a chunk is still only given up once all but one of its close group has proven it holds a copy. Both are pinned by tests, along with the fact that halving the wave spacing only ever moves a node earlier and never closes a wave that had opened. Separately, `storage_empty` in the commitment rotation meant "nothing left to commit to" rather than "no bytes anywhere". Once the migration settles, the commitment narrows to the file-backed set, so a node that could not copy anything before its disk filled reported an empty set while a full legacy store sat beside it. That took the `clear_all` branch and dropped every retained root, and an auditor holding a valid pin then got an UnknownCommitment, graded as a confirmed failure, on a node that could have answered the challenge from the store sitting next to it. --- config/production.toml | 51 +----- ...e-based-chunk-store-and-lmdb-retirement.md | 43 +++++ src/config.rs | 53 ++++-- src/replication/mod.rs | 24 ++- src/storage/chunk_store.rs | 110 ++++++++---- src/storage/migration.rs | 166 ++++++++++++++---- 6 files changed, 321 insertions(+), 126 deletions(-) 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/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/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)); } From 865ed578c4cf1ce77b6f2f43d1b287d4e90f15ac Mon Sep 17 00:00:00 2001 From: grumbach Date: Wed, 9 Sep 2026 15:16:13 +0900 Subject: [PATCH 2/2] feat(storage): say on the wire whether a node still holds an old chunk store 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; the ones most likely to still be carrying a chunks.mdb are the ones we do not. So a node says so itself, in the user agent saorsa-core already sends with every signed message and keeps for each peer. No new message, no new field, no protocol version: a different value in a string that was already on the wire. The `node/` prefix stays, because saorsa-core gates DHT membership on it and is its only consumer. Three states, never folded into two: a directory that could not be read is not one that is not there, and reading "cannot tell" as "finished" is how a gate comes back clean over a fleet that is not. Each node also reports what it sees of its peers, one line per peer that has NOT finished, at info rather than debug because nodes run at info and a line nobody emits cannot gate anything. Naming only the unfinished peers keeps the volume bounded and falls to nothing as the answer arrives; the aggregate line is emitted either way, so the denominator does not go missing and a node that has gone quiet stays distinguishable from one with nothing to report. Only a node's own line can close the gate. The user agent is fixed when the transport is built, so a node that finishes migrating goes on telling peers it has an old store until it restarts. Peer observation is how nodes we have no logs from are found at all; it can never reach zero. The reporter holds a weak handle to the node and upgrades it per pass. A reporter must not be the reason the thing it observes stays alive: a strong handle would keep a dropped node's transport, and its bound port, for as long as the task ran. Also persists the staged-rollout window under the node root. It was in memory only, so a node restarting inside its own delay restarted its own timer and could defer an upgrade indefinitely, which is the difference between "the fleet had two weeks" and "the fleet had two weeks unless it restarted". Every failure to read or write it resolves to "upgrade now", never to "keep waiting". The terraform worker unit passes --enable-logging. Without it the binary installs no subscriber and a node on that path emits nothing at all, so none of the above is readable. --- deploy/terraform/cloud-init/worker.yml | 7 +- src/node.rs | 43 +- src/storage/migration_signal.rs | 699 +++++++++++++++++++++++++ src/storage/mod.rs | 2 + src/upgrade/mod.rs | 2 + src/upgrade/monitor.rs | 99 +++- src/upgrade/rollout_state.rs | 312 +++++++++++ 7 files changed, 1145 insertions(+), 19 deletions(-) create mode 100644 src/storage/migration_signal.rs create mode 100644 src/upgrade/rollout_state.rs 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/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/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); + } +}