Repository: paritytech/parity-db (language: rust)

Issue:
After recovery on corrupted log, existing values might get overriden after unrelated commit processing
Senario:
1. The database is created with reference-counting.
2. A first commit C1 is done, inserting a key K1, the commit is processed and the log is flushed
3. A first commit C2 is done, inserting a key K2 is inserted, and the commit is processed.
4. A third commit C3 is done (might be empty), but the I/Os fail during commit processing.
5. The database is closed and open again. The second log file is invalid and removed.
6. The database is closed and open a second time. K1 is still present in the DB.
7. A fourth commit C4 is done and processed. When writing a `Key mismatch at 2` log is printed.
8. K1 is not in the database anymore.

As a test:
```rust
	#[test]
	fn test_from_fuzzer() {
		let tmp = tempdir().unwrap();
		let mut options = Options::with_columns(tmp.path(), 1);
		options.columns[0].ref_counted = true;
		options.columns[0].preimage = true;
		options.always_flush = true;
		options.with_background_thread = false;

		{
			let db = Db::open_or_create(&options).unwrap();
			db.commit_changes(vec![(0, Operation::Set(vec![59], vec![59]))]).unwrap();
			db.process_commits().unwrap();
			db.flush_logs().unwrap();
			db.commit_changes(vec![(0, Operation::Set(vec![0], vec![0]))]).unwrap();
			db.process_commits().unwrap();
			db.commit_changes(vec![]).unwrap();
			crate::set_number_of_allowed_io_operations(3);
			assert!(db.process_commits().is_err()); // Partial write
			crate::set_number_of_allowed_io_operations(usize::MAX);
		}

		{
			let db = Db::open(&options).unwrap();
			assert!(db.get(0, &[0]).unwrap().is_some()) // Key 0 is here
		}

		{
			let db = Db::open(&options).unwrap();
			assert!(db.get(0, &[0]).unwrap().is_some()); // Key 0 is here
			db.commit_changes(vec![(0, Operation::Set(vec![196], vec![196]))])
			.unwrap();
			db.process_commits().unwrap();
			assert!(db.get(0, &[0]).unwrap().is_none()) // Key 0 has been removed
		}
	}
```

Before C4 the database content is:
```
Index key: [22, 1f, f2, 5f, 13, 93, be, df, 22, cc, fd, 8e, 6b, 6b, bf, e1, bf, d6, 3f, 3c, 27, 2a, 9c, 1f, db, 53, be, a3, 18, 15, 32, e0]
        Rc: 1
Value: 3b
Index key: [f3, 71, 9b, 9e, 9f, be, 4, e3, 5f, 25, 5c, d7, a9, 29, c5, 24, d9, 8d, b1, 21, 19, 2f, 64, 6e, 52, c3, 3b, 9b, 1b, d8, 6d, 6f]
        Rc: 1
Value: 00
```

And after:
```
for 65536 chunks of column 0
Index key: [22, 1f, f2, 5f, 13, 93, be, df, 22, cc, fd, 8e, 6b, 6b, bf, e1, bf, d6, 3f, 3c, 27, 2a, 9c, 1f, db, 53, be, a3, 18, 15, 32, e0]
        Rc: 1
parity_db::column] Value: 3b
Index key: [a5, fa, ce, bb, 91, 75, b7, d7, c5, 4b, de, f6, 65, e1, a5, 2d, 82, b2, 29, 33, 2c, 60, ab, 29, ab, 57, 8, 15, b2, dc, c6, ed]
        Rc: 1
Value: c4
Index key: [f3, 71, 9b, 9e, 9f, be, b7, d7, c5, 4b, de, f6, 65, e1, a5, 2d, 82, b2, 29, 33, 2c, 60, ab, 29, ab, 57, 8, 15, b2, dc, c6, ed]
        Rc: 1
Value: c4
```

It seems like somehow the (0, 0) key-value partially overridden by the (196-196) key-value.


Discussion / hints:

Fix patch:
diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml
index cfc85775..86812026 100644
--- a/.github/workflows/rust.yml
+++ b/.github/workflows/rust.yml
@@ -17,7 +17,7 @@ jobs:
     - name: Build
       run: cargo build --verbose
     - name: Run tests
-      run: cargo test --verbose
+      run: cargo test --features instrumentation --verbose
 
   fmt:
     runs-on: ubuntu-latest
diff --git a/src/db.rs b/src/db.rs
index a90ba034..f7da2c3c 100644
--- a/src/db.rs
+++ b/src/db.rs
@@ -519,7 +519,7 @@ impl DbInner {
 				Ok(reader) => reader,
 				Err(Error::Corruption(_)) if validation_mode => {
 					log::debug!(target: "parity-db", "Bad log header");
-					self.log.clear_replay_logs()?;
+					self.log.clear_replay_logs();
 					return Ok(false)
 				},
 				Err(e) => return Err(e),
@@ -539,7 +539,7 @@ impl DbInner {
 							reader.record_id(),
 						);
 						drop(reader);
-						self.log.clear_replay_logs()?;
+						self.log.clear_replay_logs();
 						return Ok(false)
 					}
 					// Validate all records before applying anything
@@ -549,7 +549,7 @@ impl DbInner {
 							Err(e) => {
 								log::debug!(target: "parity-db", "Error reading log: {:?}", e);
 								drop(reader);
-								self.log.clear_replay_logs()?;
+								self.log.clear_replay_logs();
 								return Ok(false)
 							},
 						};
@@ -557,7 +557,7 @@ impl DbInner {
 							LogAction::BeginRecord => {
 								log::debug!(target: "parity-db", "Unexpected log header");
 								drop(reader);
-								self.log.clear_replay_logs()?;
+								self.log.clear_replay_logs();
 								return Ok(false)
 							},
 							LogAction::EndRecord => break,
@@ -574,7 +574,7 @@ impl DbInner {
 								) {
 									log::warn!(target: "parity-db", "Error replaying log: {:?}. Reverting", e);
 									drop(reader);
-									self.log.clear_replay_logs()?;
+									self.log.clear_replay_logs();
 									return Ok(false)
 								}
 							},
@@ -591,7 +591,7 @@ impl DbInner {
 								) {
 									log::warn!(target: "parity-db", "Error replaying log: {:?}. Reverting", e);
 									drop(reader);
-									self.log.clear_replay_logs()?;
+									self.log.clear_replay_logs();
 									return Ok(false)
 								}
 							},
@@ -741,7 +741,8 @@ impl DbInner {
 				// On error the log reader may be left in inconsistent state. So it is important
 				// to no attempt any further log enactment.
 				log::debug!(target: "parity-db", "Shutdown with error state {}", err);
-				return Ok(())
+				self.log.clean_logs(self.log.num_dirty_logs())?;
+				return self.log.kill_logs()
 			}
 		}
 		log::debug!(target: "parity-db", "Processing leftover commits");
@@ -851,7 +852,12 @@ impl Db {
 		let mut db = DbInner::open(options, opening_mode)?;
 		// This needs to be call before log thread: so first reindexing
 		// will run in correct state.
-		db.replay_all_logs()?;
+		if let Err(e) = db.replay_all_logs() {
+			log::debug!(target: "parity-db", "Error during log replay, doing log cleanup");
+			db.log.clean_logs(db.log.num_dirty_logs())?;
+			db.log.kill_logs()?;
+			return Err(e)
+		}
 		let db = Arc::new(db);
 		#[cfg(any(test, feature = "instrumentation"))]
 		let start_threads = opening_mode != OpeningMode::ReadOnly && options.with_background_thread;
@@ -2381,4 +2387,38 @@ mod tests {
 			assert_eq!(iter_state_rev.next(), next.as_ref().map(|(k, v)| (k, v)));
 		}
 	}
+
+	#[cfg(feature = "instrumentation")]
+	#[test]
+	fn test_partial_log_recovery() {
+		let tmp = tempdir().unwrap();
+		let mut options = Options::with_columns(tmp.path(), 1);
+		options.columns[0].btree_index = true;
+		options.always_flush = true;
+		options.with_background_thread = false;
+
+		// We do 2 commits and we fail while writing the second one
+		{
+			let db = Db::open_or_create(&options).unwrap();
+			db.commit::<_, Vec<u8>>(vec![(0, vec![0], Some(vec![0]))]).unwrap();
+			db.process_commits().unwrap();
+			db.commit::<_, Vec<u8>>(vec![(0, vec![1], Some(vec![1]))]).unwrap();
+			crate::set_number_of_allowed_io_operations(4);
+			assert!(db.process_commits().is_err());
+			crate::set_number_of_allowed_io_operations(usize::MAX);
+			db.flush_logs().unwrap();
+		}
+
+		// We open a first time, the first value is there
+		{
+			let db = Db::open(&options).unwrap();
+			assert_eq!(db.get(0, &[0]).unwrap(), Some(vec![0]));
+		}
+
+		// We open a second time, the first value should be still there
+		{
+			let db = Db::open(&options).unwrap();
+			assert!(db.get(0, &[0]).unwrap().is_some());
+		}
+	}
 }
diff --git a/src/log.rs b/src/log.rs
index 8e1b40b5..7433d214 100644
--- a/src/log.rs
+++ b/src/log.rs
@@ -580,21 +580,12 @@ impl Log {
 		Ok(())
 	}
 
-	pub fn clear_replay_logs(&self) -> Result<()> {
-		{
-			let mut reading = self.reading.write();
-			let id = reading.as_ref().map(|r| r.id);
-			*reading = None;
-			if let Some(id) = id {
-				self.drop_log(id)?;
-			}
+	pub fn clear_replay_logs(&self) {
+		if let Some(reading) = self.reading.write().take() {
+			self.cleanup_queue.write().push_back((reading.id, reading.file.into_inner()));
 		}
-		{
-			let replay_logs = std::mem::take(&mut *self.replay_queue.write());
-			for (id, _, file) in replay_logs {
-				drop(file);
-				self.drop_log(id)?;
-			}
+		for (id, _, file) in self.replay_queue.write().drain(0..) {
+			self.cleanup_queue.write().push_back((id, file));
 		}
 		let mut overlays = self.overlays.write();
 		overlays.index.clear();
@@ -602,7 +593,6 @@ impl Log {
 		overlays.last_record_id.clear();
 		*self.reading_state.lock() = ReadingState::Idle;
 		self.dirty.store(false, Ordering::Relaxed);
-		Ok(())
 	}
 
 	pub fn begin_record(&self) -> LogWriter<'_> {


Test patch: