Method upsert

nodedb-cluster/src/metadata_group/migration_state.rs:246–279

Upsert a checkpoint row (in-memory + redb). Idempotent: if the exact `(migration_id, phase, attempt)` tuple is already stored, this is a no-op.

(&mut self, row: PersistedMigrationCheckpoint)

    /// Idempotent: if the exact `(migration_id, phase, attempt)` tuple
    /// is already stored, this is a no-op.
    pub fn upsert(&mut self, row: PersistedMigrationCheckpoint) -> Result<()> {
        let key = row.migration_id.clone();
        // Idempotency: skip if identical (migration_id, phase_tag, attempt) already stored.
        if let Some(existing) = self.mem.get(&key)
            && existing.payload.phase_tag() == row.payload.phase_tag()
            && existing.attempt == row.attempt
        {
            return Ok(());
        }
        // Encode and persist.
        let bytes = zerompk::to_msgpack_vec(&row).map_err(|e| ClusterError::Codec {
            detail: format!("migration_state encode: {e}"),
        })?;
        let txn = self.db.begin_write().map_err(|e| ClusterError::Storage {
            detail: format!("migration_state begin_write: {e}"),
        })?;
        {
            let mut table = txn
                .open_table(Self::TABLE)
                .map_err(|e| ClusterError::Storage {
                    detail: format!("migration_state open_table: {e}"),
                })?;
            table
                .insert(key.as_str(), bytes.as_slice())
                .map_err(|e| ClusterError::Storage {
                    detail: format!("migration_state insert: {e}"),
                })?;
        }
        txn.commit().map_err(|e| ClusterError::Storage {
            detail: format!("migration_state commit: {e}"),
        })?;
        self.mem.insert(key, row);
        Ok(())
    }

    /// Remove a checkpoint row (in-memory + redb).  No-op if not present.
    pub fn remove(&mut self, migration_id: &MigrationId) -> Result<()> {