@@ -818,6 +818,13 @@ impl Runtime { |
818 | 818 | return; |
819 | 819 | } |
820 | 820 | } |
| 821 | + #[cfg(any(test, feature = "testing"))] |
| 822 | + Ok(Command::InjectFatalError) => { |
| 823 | + let error = DbspError::IO(std::io::Error::other("injected fatal error")); |
| 824 | + if status_sender.send(Err(error)).is_err() { |
| 825 | + return; |
| 826 | + } |
| 827 | + } |
821 | 828 | Ok(Command::Rebalance) => { |
822 | 829 | circuit.rebalance(); |
823 | 830 | if status_sender.send(Ok(Response::Unit)).is_err() { |
@@ -1151,6 +1158,58 @@ enum Command { |
1151 | 1158 | /// Install the bootstrap circuit's state into the main circuit, |
1152 | 1159 | /// reactivate the full main circuit, and delete the bootstrap circuit. |
1153 | 1160 | CutoverBootstrapCircuit, |
| 1161 | + /// Test-only: every worker replies with an error that is fatal. |
| 1162 | + #[cfg(any(test, feature = "testing"))] |
| 1163 | + InjectFatalError, |
| 1164 | +} |
| 1165 | + |
| 1166 | +impl Command { |
| 1167 | + /// Returns true if a worker's error reply to this command can leave the |
| 1168 | + /// workers out of step, so that the runtime must be killed. |
| 1169 | + /// |
| 1170 | + /// Returns false only for commands that do not change circuit state, so |
| 1171 | + /// that the circuit stays usable after an error. |
| 1172 | + fn error_is_fatal(&self) -> bool { |
| 1173 | + match self { |
| 1174 | + // Takes in-memory snapshots of operator state. It writes no files |
| 1175 | + // and does no exchange between workers. |
| 1176 | + Command::PrepareCheckpoint => false, |
| 1177 | + |
| 1178 | + Command::StartTransaction |
| 1179 | + | Command::Step |
| 1180 | + | Command::CommitTransaction |
| 1181 | + | Command::CommitProgress |
| 1182 | + | Command::Transaction |
| 1183 | + | Command::IsBootstrapComplete |
| 1184 | + | Command::CompleteBootstrap |
| 1185 | + | Command::EnableProfiler |
| 1186 | + | Command::DumpProfile { .. } |
| 1187 | + | Command::RetrieveGraph |
| 1188 | + | Command::RetrieveProfile { .. } |
| 1189 | + | Command::GetLir |
| 1190 | + | Command::Restore(_) |
| 1191 | + | Command::SetBalancerHintsByGlobalId(_) |
| 1192 | + | Command::SetBalancerHints(_) |
| 1193 | + | Command::GetCurrentBalancerPolicies |
| 1194 | + | Command::GetCurrentBalancerPolicy(_) |
| 1195 | + | Command::Rebalance |
| 1196 | + | Command::SetAutoRebalance(_) |
| 1197 | + | Command::StartCompaction |
| 1198 | + | Command::IsCompactionComplete |
| 1199 | + | Command::CreateBootstrapCircuit(_) |
| 1200 | + | Command::StartBootstrapTransaction |
| 1201 | + | Command::CommitBootstrapTransaction |
| 1202 | + | Command::StepBootstrapCircuit |
| 1203 | + | Command::BootstrapCommitProgress |
| 1204 | + | Command::DestroyBootstrapCircuit |
| 1205 | + | Command::RestoreConcurrent(_) |
| 1206 | + | Command::SyncBootstrapCircuit |
| 1207 | + | Command::CutoverBootstrapCircuit => true, |
| 1208 | + |
| 1209 | + #[cfg(any(test, feature = "testing"))] |
| 1210 | + Command::InjectFatalError => true, |
| 1211 | + } |
| 1212 | + } |
1154 | 1213 | } |
1155 | 1214 |
|
1156 | 1215 | impl Debug for Command { |
@@ -1208,6 +1267,8 @@ impl Debug for Command { |
1208 | 1267 | } |
1209 | 1268 | Command::SyncBootstrapCircuit => write!(f, "SyncBootstrapCircuit"), |
1210 | 1269 | Command::CutoverBootstrapCircuit => write!(f, "CutoverBootstrapCircuit"), |
| 1270 | + #[cfg(any(test, feature = "testing"))] |
| 1271 | + Command::InjectFatalError => write!(f, "InjectFatalError"), |
1211 | 1272 | } |
1212 | 1273 | } |
1213 | 1274 | } |
@@ -1433,7 +1494,9 @@ impl DBSPHandle { |
1433 | 1494 | Err(DbspError::Runtime(RuntimeError::WorkerPanic { panic_info })) |
1434 | 1495 | } |
1435 | 1496 |
|
1436 | | - // Receive responses. |
| 1497 | + // Receive responses. After a non-fatal error, keep receiving, so that |
| 1498 | + // no stale reply is left for the next command. |
| 1499 | + let mut error = None; |
1437 | 1500 | for _ in 0..self.status_receivers.len() { |
1438 | 1501 | let ready = select.select(); |
1439 | 1502 | let worker = ready.index(); |
@@ -1445,18 +1508,21 @@ impl DBSPHandle { |
1445 | 1508 |
|
1446 | 1509 | match ready.recv(&self.status_receivers[worker]) { |
1447 | 1510 | Err(_) => return handle_panic(self), |
1448 | | - Ok(Err(e)) => { |
| 1511 | + Ok(Err(e)) if command.error_is_fatal() => { |
1449 | 1512 | let _ = self.kill_inner(); |
1450 | 1513 | return Err(e); |
1451 | 1514 | } |
| 1515 | + Ok(Err(e)) => { |
| 1516 | + error.get_or_insert(e); |
| 1517 | + } |
1452 | 1518 | Ok(Ok(resp)) => handler(worker, resp), |
1453 | 1519 | } |
1454 | 1520 | } |
1455 | 1521 | if self.panicked() { |
1456 | 1522 | return handle_panic(self); |
1457 | 1523 | } |
1458 | 1524 |
|
1459 | | - Ok(()) |
| 1525 | + error.map_or(Ok(()), Err) |
1460 | 1526 | } |
1461 | 1527 |
|
1462 | 1528 | fn unicast_command(&mut self, worker: usize, command: Command) -> Result<Response, DbspError> { |
@@ -1465,6 +1531,7 @@ impl DBSPHandle { |
1465 | 1531 | } |
1466 | 1532 |
|
1467 | 1533 | // Send command. |
| 1534 | + let error_is_fatal = command.error_is_fatal(); |
1468 | 1535 | if self.command_senders[worker].send(command).is_err() { |
1469 | 1536 | let panic_info = self.collect_panic_info().unwrap_or_default(); |
1470 | 1537 |
|
@@ -1488,10 +1555,11 @@ impl DBSPHandle { |
1488 | 1555 | } |
1489 | 1556 | let reply = match ready.recv(&self.status_receivers[worker]) { |
1490 | 1557 | Err(_) => return handle_panic(self), |
1491 | | - Ok(Err(e)) => { |
| 1558 | + Ok(Err(e)) if error_is_fatal => { |
1492 | 1559 | let _ = self.kill_inner(); |
1493 | 1560 | return Err(e); |
1494 | 1561 | } |
| 1562 | + Ok(Err(e)) => return Err(e), |
1495 | 1563 | Ok(Ok(resp)) => resp, |
1496 | 1564 | }; |
1497 | 1565 |
|
@@ -2414,6 +2482,14 @@ impl DBSPHandle { |
2414 | 2482 | Ok(()) |
2415 | 2483 | } |
2416 | 2484 |
|
| 2485 | + /// Test-only: makes every worker fail a command with an error that is |
| 2486 | + /// fatal, so that the runtime is killed from the calling thread, as it is |
| 2487 | + /// for a real fatal error. |
| 2488 | + #[cfg(any(test, feature = "testing"))] |
| 2489 | + pub fn inject_fatal_error(&mut self) -> Result<(), DbspError> { |
| 2490 | + self.broadcast_command(Command::InjectFatalError, |_, _| {}) |
| 2491 | + } |
| 2492 | + |
2417 | 2493 | pub fn start_compaction(&mut self) -> Result<(), DbspError> { |
2418 | 2494 | self.broadcast_command(Command::StartCompaction, |_, _| {})?; |
2419 | 2495 | Ok(()) |
@@ -3237,6 +3313,67 @@ pub(crate) mod tests { |
3237 | 3313 | } |
3238 | 3314 | } |
3239 | 3315 |
|
| 3316 | + /// A failed checkpoint must return the error and leave the circuit |
| 3317 | + /// running. Only worker 0 fails, so the other workers' replies must be |
| 3318 | + /// drained, not left for the next command. |
| 3319 | + #[test] |
| 3320 | + fn test_failed_checkpoint_keeps_circuit() { |
| 3321 | + use crate::circuit::Mode; |
| 3322 | + use crate::typed_batch::IndexedZSetReader; |
| 3323 | + use std::sync::atomic::{AtomicI64, Ordering}; |
| 3324 | + |
| 3325 | + let temp = tempdir().unwrap(); |
| 3326 | + let total = Arc::new(AtomicI64::new(0)); |
| 3327 | + let total_clone = total.clone(); |
| 3328 | + let cconf = mkconfig_with(temp.path(), 4, 0).with_mode(Mode::Persistent); |
| 3329 | + let (mut dbsp, input) = Runtime::init_circuit(cconf, move |circuit| { |
| 3330 | + let (stream, handle) = circuit.add_input_zset::<i32>(); |
| 3331 | + stream.set_persistent_id(Some("input")); |
| 3332 | + let total = total_clone.clone(); |
| 3333 | + stream.inspect(move |batch: &OrdZSet<i32>| { |
| 3334 | + total.fetch_add(batch.iter().map(|(_, _, w)| w).sum(), Ordering::SeqCst); |
| 3335 | + }); |
| 3336 | + if Runtime::worker_index() == 0 { |
| 3337 | + // No persistent id, so this operator cannot be checkpointed. |
| 3338 | + stream.integrate(); |
| 3339 | + } |
| 3340 | + Ok(handle) |
| 3341 | + }) |
| 3342 | + .unwrap(); |
| 3343 | + |
| 3344 | + for round in 1..=2 { |
| 3345 | + let err = dbsp.checkpoint().run().unwrap_err(); |
| 3346 | + assert!( |
| 3347 | + matches!(err, DbspError::Storage(StorageError::NoPersistentId(_))), |
| 3348 | + "unexpected error {err:?}" |
| 3349 | + ); |
| 3350 | + |
| 3351 | + // Panics on a stale `CheckpointPrepared` reply. |
| 3352 | + dbsp.commit_progress().unwrap(); |
| 3353 | + |
| 3354 | + input.append(&mut vec![Tup2(round, 1), Tup2(round + 10, 1)]); |
| 3355 | + dbsp.transaction().unwrap(); |
| 3356 | + assert_eq!(total.load(Ordering::SeqCst), 2 * round as i64); |
| 3357 | + } |
| 3358 | + dbsp.kill().unwrap(); |
| 3359 | + } |
| 3360 | + |
| 3361 | + /// A fatal error kills the runtime, so later commands fail. |
| 3362 | + #[test] |
| 3363 | + fn test_fatal_error_kills_circuit() { |
| 3364 | + let temp = tempdir().unwrap(); |
| 3365 | + let (mut dbsp, ()) = |
| 3366 | + Runtime::init_circuit(mkconfig_with(temp.path(), 4, 0), |_circuit| Ok(())).unwrap(); |
| 3367 | + |
| 3368 | + let err = dbsp.inject_fatal_error().unwrap_err(); |
| 3369 | + assert!(matches!(err, DbspError::IO(_)), "unexpected error {err:?}"); |
| 3370 | + let err = dbsp.transaction().unwrap_err(); |
| 3371 | + assert!( |
| 3372 | + matches!(err, DbspError::Runtime(RuntimeError::Terminated)), |
| 3373 | + "unexpected error {err:?}" |
| 3374 | + ); |
| 3375 | + } |
| 3376 | + |
3240 | 3377 | type CircuitHandle = ( |
3241 | 3378 | IndexedZSetHandle<i32, i32>, |
3242 | 3379 | OutputHandle<OrdZSet<Tup2<i32, i32>>>, |
|