Shift the version from State to StateCollections

Author: bkirwi

src/persist-client/src/cli/admin.rs

@@ -590,9 +590,9 @@ where
         // code with this logic.
         let safe_version_change = match (commit, expected_version) {
             // We never actually write out state changes, so increasing the version is okay.
-            (false, _) => cfg.build_version >= state.applier_version,
+            (false, _) => cfg.build_version >= state.collections.version,
             // If the versions match that's okay because any commits won't change it.
-            (true, None) => cfg.build_version == state.applier_version,
+            (true, None) => cfg.build_version == state.collections.version,
             // !!DANGER ZONE!!
             (true, Some(expected)) => {
                 // If we're not _extremely_ careful, the persistcli could make shards unreadable by
@@ -602,15 +602,15 @@ where
                 // We only allow a mismatch in version if we provided the expected version to the
                 // command, and the expected version is less than the current build, which
                 // indicates this is an old shard.
-                state.applier_version == expected && expected <= cfg.build_version
+                state.collections.version == expected && expected <= cfg.build_version
             }
         };
         if !safe_version_change {
             // We could add a flag to override this check, if that comes up.
             return Err(anyhow!(
                 "version of this tool {} does not match version of state {} when --commit is {commit}. bailing so we don't corrupt anything",
                 cfg.build_version,
-                state.applier_version
+                state.collections.version
             ));
         }
         break;

src/persist-client/src/internal/encoding.rs

@@ -736,7 +736,7 @@ impl<T: Timestamp + Lattice + Codec64> UntypedState<T> {
         let state = Rollup::from_proto(proto)
             .expect("internal error: invalid encoded state")
             .state;
-        check_data_version(build_version, &state.state.applier_version);
+        check_data_version(build_version, &state.state.collections.version);
         state
     }
 }
@@ -864,7 +864,7 @@ impl RustType<ProtoInlinedDiffs> for InlinedDiffs {
 impl<T: Timestamp + Lattice + Codec64> RustType<ProtoRollup> for Rollup<T> {
     fn into_proto(&self) -> ProtoRollup {
         ProtoRollup {
-            applier_version: self.state.state.applier_version.to_string(),
+            applier_version: self.state.state.collections.version.to_string(),
             shard_id: self.state.state.shard_id.into_proto(),
             seqno: self.state.state.seqno.into_proto(),
             walltime_ms: self.state.state.walltime_ms.into_proto(),
@@ -972,6 +972,7 @@ impl<T: Timestamp + Lattice + Codec64> RustType<ProtoRollup> for Rollup<T> {
             .transpose()?;
         let active_gc = x.active_gc.map(|gc| gc.into_rust()).transpose()?;
         let collections = StateCollections {
+            version: applier_version.clone(),
             rollups,
             active_rollup,
             active_gc,
@@ -983,7 +984,6 @@ impl<T: Timestamp + Lattice + Codec64> RustType<ProtoRollup> for Rollup<T> {
             trace: x.trace.into_rust_if_some("trace")?,
         };
         let state = State {
-            applier_version,
             shard_id: x.shard_id.into_rust()?,
             seqno: x.seqno.into_rust()?,
             walltime_ms: x.walltime_ms,

src/persist-client/src/internal/state.rs

@@ -1304,6 +1304,11 @@ pub struct NoOpStateTransition<T>(pub T);
 #[derive(Debug, Clone)]
 #[cfg_attr(any(test, debug_assertions), derive(PartialEq))]
 pub struct StateCollections<T> {
+    /// The version of this state. This is typically identical to the version of the code
+    /// that wrote it, but may diverge during 0dt upgrades and similar operations when a
+    /// new version of code is intentionally interoperating with an older state format.
+    pub(crate) version: Version,
+
     // - Invariant: `<= all reader.since`
     // - Invariant: Doesn't regress across state versions.
     pub(crate) last_gc_req: SeqNo,
@@ -2221,10 +2226,6 @@ where
 #[derive(Debug)]
 #[cfg_attr(any(test, debug_assertions), derive(Clone, PartialEq))]
 pub struct State<T> {
-    /// The version of this state. This is typically identical to the version of the code
-    /// that wrote it, but may diverge during 0dt upgrades and similar operations when a
-    /// new version of code is intentionally interoperating with an older state format.
-    pub(crate) applier_version: semver::Version,
     pub(crate) shard_id: ShardId,
 
     pub(crate) seqno: SeqNo,
@@ -2254,10 +2255,9 @@ pub struct TypedState<K, V, T, D> {
 
 impl<K, V, T: Clone, D> TypedState<K, V, T, D> {
     #[cfg(any(test, debug_assertions))]
-    pub(crate) fn clone(&self, applier_version: Version, hostname: String) -> Self {
+    pub(crate) fn clone(&self, hostname: String) -> Self {
         TypedState {
             state: State {
-                applier_version,
                 shard_id: self.shard_id.clone(),
                 seqno: self.seqno.clone(),
                 walltime_ms: self.walltime_ms,
@@ -2271,7 +2271,6 @@ impl<K, V, T: Clone, D> TypedState<K, V, T, D> {
     pub(crate) fn clone_for_rollup(&self) -> Self {
         TypedState {
             state: State {
-                applier_version: self.applier_version.clone(),
                 shard_id: self.shard_id.clone(),
                 seqno: self.seqno.clone(),
                 walltime_ms: self.walltime_ms,
@@ -2338,12 +2337,12 @@ where
         walltime_ms: u64,
     ) -> Self {
         let state = State {
-            applier_version,
             shard_id,
             seqno: SeqNo::minimum(),
             walltime_ms,
             hostname,
             collections: StateCollections {
+                version: applier_version,
                 last_gc_req: SeqNo::minimum(),
                 rollups: BTreeMap::new(),
                 active_rollup: None,
@@ -2373,15 +2372,16 @@ where
         // each version of state with the _max_ version of code that has ever
         // contributed to it. Otherwise, we'd erroneously allow rolling back an
         // arbitrary number of versions if they were done one-by-one.
-        let new_applier_version = std::cmp::max(&self.applier_version, &cfg.build_version);
+        let new_applier_version = std::cmp::max(&self.collections.version, &cfg.build_version);
         let mut new_state = State {
-            applier_version: new_applier_version.clone(),
             shard_id: self.shard_id,
             seqno: self.seqno.next(),
             walltime_ms: (cfg.now)(),
             hostname: cfg.hostname.clone(),
             collections: self.collections.clone(),
         };
+        new_state.collections.version = new_applier_version.clone();
+
         // Make sure walltime_ms is strictly increasing, in case clocks are
         // offset.
         if new_state.walltime_ms <= self.walltime_ms {
@@ -2752,13 +2752,13 @@ fn serialize_diffs_sum<S: Serializer>(val: &Option<[u8; 8]>, s: S) -> Result<S::
 impl<T: Serialize + Timestamp + Lattice> Serialize for State<T> {
     fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
         let State {
-            applier_version,
             shard_id,
             seqno,
             walltime_ms,
             hostname,
             collections:
                 StateCollections {
+                    version: applier_version,
                     last_gc_req,
                     rollups,
                     active_rollup,
@@ -3140,12 +3140,12 @@ pub(crate) mod tests {
                 (shard_id, seqno, walltime_ms, hostname, last_gc_req, rollups, active_rollup),
                 (active_gc, leased_readers, critical_readers, writers, schemas, trace),
             )| State {
-                applier_version: semver::Version::new(1, 2, 3),
                 shard_id,
                 seqno,
                 walltime_ms,
                 hostname,
                 collections: StateCollections {
+                    version: Version::new(1, 2, 3),
                     last_gc_req,
                     rollups,
                     active_rollup,

src/persist-client/src/internal/state_diff.rs

@@ -146,13 +146,13 @@ impl<T: Timestamp + Lattice + Codec64> StateDiff<T> {
         // Deconstruct from and to so we get a compile failure if new
         // fields are added.
         let State {
-            applier_version: _,
             shard_id: from_shard_id,
             seqno: from_seqno,
             hostname: from_hostname,
             walltime_ms: _, // Intentionally unused
             collections:
                 StateCollections {
+                    version: _,
                     last_gc_req: from_last_gc_req,
                     rollups: from_rollups,
                     active_rollup: from_active_rollup,
@@ -165,13 +165,13 @@ impl<T: Timestamp + Lattice + Codec64> StateDiff<T> {
                 },
         } = from;
         let State {
-            applier_version: to_applier_version,
             shard_id: to_shard_id,
             seqno: to_seqno,
             walltime_ms: to_walltime_ms,
             hostname: to_hostname,
             collections:
                 StateCollections {
+                    version: to_applier_version,
                     last_gc_req: to_last_gc_req,
                     rollups: to_rollups,
                     active_rollup: to_active_rollup,
@@ -319,10 +319,7 @@ impl<T: Timestamp + Lattice + Codec64> StateDiff<T> {
 
         use crate::internal::state::ProtoStateDiff;
 
-        let mut roundtrip_state = from_state.clone(
-            from_state.applier_version.clone(),
-            from_state.hostname.clone(),
-        );
+        let mut roundtrip_state = from_state.clone(from_state.hostname.clone());
         roundtrip_state.apply_diff(metrics, diff.clone())?;
 
         if &roundtrip_state != to_state {
@@ -397,7 +394,8 @@ impl<T: Timestamp + Lattice + Codec64> State<T> {
                     // issues that may arise from diff application. We pass along the original
                     // Bytes it decoded from just so we can decode in this error path, while
                     // avoiding any extraneous clones in the expected Ok path.
-                    let diff = StateDiff::<T>::decode(&self.applier_version, data);
+                    // FIXME: this passes the state version but the method requires the build version.
+                    let diff = StateDiff::<T>::decode(&self.collections.version, data);
                     panic!(
                         "state diff should apply cleanly: {} diff {:?} state {:?}",
                         err, diff, self
@@ -446,7 +444,6 @@ impl<T: Timestamp + Lattice + Codec64> State<T> {
             ));
         }
         self.seqno = diff_seqno_to;
-        self.applier_version = diff_applier_version;
         self.walltime_ms = diff_walltime_ms;
         force_apply_diffs_single(
             &self.shard_id,
@@ -460,6 +457,7 @@ impl<T: Timestamp + Lattice + Codec64> State<T> {
         // Deconstruct collections so we get a compile failure if new fields are
         // added.
         let StateCollections {
+            version,
             last_gc_req,
             rollups,
             active_rollup,
@@ -471,6 +469,7 @@ impl<T: Timestamp + Lattice + Codec64> State<T> {
             trace,
         } = &mut self.collections;
 
+        *version = diff_applier_version;
         apply_diffs_map("rollups", diff_rollups, rollups)?;
         apply_diffs_single("last_gc_req", diff_last_gc_req, last_gc_req)?;
         apply_diffs_single_option("active_rollup", diff_active_rollup, active_rollup)?;

src/persist-client/src/internal/state_versions.rs

@@ -1074,7 +1074,6 @@ impl<T: Timestamp + Lattice + Codec64> StateVersionsIter<T> {
     pub fn into_rollup_proto_without_diffs(&self) -> impl serde::Serialize + use<T> {
         Rollup::from_state_without_diffs(
             State {
-                applier_version: self.state.applier_version.clone(),
                 shard_id: self.state.shard_id.clone(),
                 seqno: self.state.seqno.clone(),
                 walltime_ms: self.state.walltime_ms.clone(),