core: Replay database prefix

If a commit cannot be decoded from the log, stop replaying and truncate
the log to the known-good prefix.

crates/core/src/db/relational_db.rs

@@ -23,6 +23,7 @@ use spacetimedb_primitives::{ColId, IndexId, SequenceId, TableId};
 use spacetimedb_sats::{AlgebraicType, AlgebraicValue, ProductType, ProductValue};
 use std::borrow::Cow;
 use std::fs::{create_dir_all, File};
+use std::io;
 use std::ops::RangeBounds;
 use std::path::Path;
 use std::sync::{Arc, Mutex};
@@ -95,10 +96,18 @@ impl RelationalDB {
             let mut last_commit_offset = None;
             let mut last_hash: Option<Hash> = None;
             if let Some(message_log) = &message_log {
-                let message_log = message_log.lock().unwrap();
+                let mut message_log = message_log.lock().unwrap();
                 let max_offset = message_log.open_segment_max_offset;
-                for commit in commit_log::Iter::from(message_log.segments()) {
-                    let commit = commit?;
+                'commits: for commit in commit_log::Iter::from(message_log.segments()) {
+                    let commit = match commit {
+                        Ok(commit) => commit,
+                        Err(e) if e.kind() == io::ErrorKind::InvalidData => {
+                            log::warn!("Corrupt commit after offset {last_commit_offset:?}");
+                            message_log.reset_to(last_commit_offset.unwrap_or(0))?;
+                            break 'commits;
+                        }
+                        Err(e) => return Err(e.into()),
+                    };
 
                     segment_index += 1;
                     last_hash = commit.parent_commit_hash;