use super::receipt_wire::{DatabaseErrorKind, WireOutcome, WireReceipt};
use haematite::{BatchFailure, BatchJobOutcome, BatchOutcome, BatchReceipt, DatabaseError, Hash};
use std::error::Error;
use std::sync::Arc;
type TestResult = Result<(), Box<dyn Error>>;
fn mixed() -> Result<BatchReceipt, Box<dyn Error>> {
let callback = Arc::new(DatabaseError::ShardError("callback cause".repeat(4096)));
let fence = Arc::new(DatabaseError::Fenced {
required: 3,
possible_accepts: 1,
});
let receipt = BatchReceipt::new(
vec![
BatchJobOutcome {
job_index: 0,
shard: 2,
outcome: BatchOutcome::Committed {
root: Hash::from_bytes([231; 32]),
callback_failure: Some(callback),
},
},
BatchJobOutcome {
job_index: 1,
shard: 3,
outcome: BatchOutcome::Failed { cause: fence },
},
BatchJobOutcome {
job_index: 2,
shard: 4,
outcome: BatchOutcome::Unknown {
cause: Arc::new(DatabaseError::BatchJobUnknown {
job_index: 2,
shard: 4,
command: "put",
}),
},
},
BatchJobOutcome {
job_index: 3,
shard: 5,
outcome: BatchOutcome::SequenceRead {
sequences: vec![(vec![0, 255, 128], u64::MAX)],
},
},
BatchJobOutcome {
job_index: 4,
shard: 6,
outcome: BatchOutcome::RangeRead {
sequences: vec![(vec![254, 0], 19)],
hit_limit: true,
invalid_outside_range: 7,
},
},
],
Some(Arc::new(DatabaseError::ExecutorShutdown)),
)?;
Ok(receipt)
}
#[test]
fn mixed_projection_preserves_every_outcome_and_original_cause() -> TestResult {
let receipt = mixed()?;
let projected = WireReceipt::project(&receipt)?;
assert_eq!(projected.outcomes.len(), 5);
let WireOutcome::Committed {
root,
callback_failure: Some(cause),
} = &projected.outcomes[0].outcome
else {
return Err("committed callback failure missing".into());
};
assert_eq!(*root, [231; 32]);
assert_eq!(
cause.message,
"shard operation failed: ".to_owned() + &"callback cause".repeat(4096)
);
assert_eq!(cause.kind, DatabaseErrorKind::ShardError);
let WireOutcome::Failed { cause } = &projected.outcomes[1].outcome else {
return Err("fenced failure missing".into());
};
assert_eq!(cause.kind, DatabaseErrorKind::Fenced);
let first_error = projected.first_error.as_ref().ok_or("no first error")?;
assert_eq!(first_error.kind, DatabaseErrorKind::ShardError);
let submit_error = projected.submit_error.as_ref().ok_or("no submit error")?;
assert_eq!(submit_error.kind, DatabaseErrorKind::ExecutorShutdown);
let WireOutcome::SequenceRead { sequences } = &projected.outcomes[3].outcome else {
return Err("sequence result missing".into());
};
assert_eq!(sequences[0].key_hex, "00ff80");
assert_eq!(sequences[0].sequence, u64::MAX);
let WireOutcome::RangeRead {
sequences,
hit_limit,
invalid_outside_range,
} = &projected.outcomes[4].outcome
else {
return Err("range result missing".into());
};
assert_eq!(sequences[0].key_hex, "fe00");
assert!(*hit_limit);
assert_eq!(*invalid_outside_range, 7);
let BatchOutcome::Committed {
callback_failure: Some(original),
..
} = &receipt.outcomes()[0].outcome
else {
return Err("original cause lost".into());
};
assert_eq!(Arc::strong_count(original), 1);
let bytes = serde_json::to_vec(&projected)?;
let decoded: WireReceipt = serde_json::from_slice(&bytes)?;
assert_eq!(decoded, projected);
Ok(())
}
#[test]
fn nested_batch_keeps_whole_receipt_and_typed_first_error() -> TestResult {
let failure = BatchFailure::new(mixed()?)?;
let receipt = BatchReceipt::new(
vec![BatchJobOutcome {
job_index: 8,
shard: 9,
outcome: BatchOutcome::Failed {
cause: Arc::new(DatabaseError::BatchFailed(Box::new(failure))),
},
}],
None,
)?;
let projected = WireReceipt::project(&receipt)?;
let WireOutcome::Failed { cause } = &projected.outcomes[0].outcome else {
return Err("nested failure missing".into());
};
assert_eq!(cause.kind, DatabaseErrorKind::BatchFailed);
let nested = cause.batch_receipt.as_ref().ok_or("no nested receipt")?;
assert_eq!(nested.outcomes.len(), 5);
let nested_first = cause.first_error.as_ref().ok_or("no nested first error")?;
assert_eq!(nested_first.kind, DatabaseErrorKind::ShardError);
let first = projected.first_error.as_ref().ok_or("no first error")?;
assert_eq!(first.kind, DatabaseErrorKind::ShardError);
Ok(())
}
pub(super) fn fixture() -> Result<BatchReceipt, Box<dyn Error>> {
mixed()
}
#[test]
fn original_batch_shard_cause_has_a_named_subkind_and_full_message() -> TestResult {
use super::receipt_wire::{ShardErrorKind, WireCause};
use haematite::ShardError;
let errors = [
(
ShardError::QueueFull {
shard: 7,
queued_bytes: 11,
request_bytes: 13,
bound: 17,
},
ShardErrorKind::QueueFull,
),
(
ShardError::RequestSizeOverflow { shard: 9 },
ShardErrorKind::RequestSizeOverflow,
),
(
ShardError::ActorUnavailable { pid: 23 },
ShardErrorKind::ActorUnavailable,
),
(
ShardError::ReplyDisconnected { pid: 29 },
ShardErrorKind::ReplyDisconnected,
),
(
ShardError::OutcomeUnknown { pid: 31 },
ShardErrorKind::OutcomeUnknown,
),
(
ShardError::Spawn("original spawn cause".repeat(4096)),
ShardErrorKind::Spawn,
),
(
ShardError::ReclaimClose {
reason: "original reclaim cause".to_owned(),
},
ShardErrorKind::ReclaimClose,
),
(
ShardError::SequenceConflict {
expected: 37,
actual: 41,
},
ShardErrorKind::SequenceConflict,
),
(
ShardError::WriteDuringRootAdvanceEmission,
ShardErrorKind::WriteDuringRootAdvanceEmission,
),
];
for (original, kind) in errors {
let message = original.to_string();
let error = DatabaseError::BatchShardFailed { cause: original };
let projected = WireCause::project(&error)?;
assert_eq!(projected.kind, DatabaseErrorKind::BatchShardFailed);
assert_eq!(projected.message, error.to_string());
let shard = projected.shard.as_ref().ok_or("shard cause missing")?;
assert_eq!(shard.kind, kind);
assert_eq!(shard.message, message);
let bytes = serde_json::to_vec(&projected)?;
let decoded: WireCause = serde_json::from_slice(&bytes)?;
assert_eq!(decoded, projected);
}
Ok(())
}
#[test]
fn batch_shard_cause_survives_nested_receipt_and_bounded_frames() -> TestResult {
use super::receipt_fragments::{Assembler, EncodedReceipt, minimum_cap};
use super::receipt_wire::ShardErrorKind;
use haematite::ShardError;
let cause = Arc::new(DatabaseError::BatchShardFailed {
cause: ShardError::Spawn("complete original shard cause".repeat(4096)),
});
let receipt = BatchReceipt::new(
vec![BatchJobOutcome {
job_index: 3,
shard: 5,
outcome: BatchOutcome::Failed {
cause: Arc::clone(&cause),
},
}],
None,
)?;
let failure = BatchFailure::new(receipt)?;
let nested = BatchReceipt::new(
vec![BatchJobOutcome {
job_index: 7,
shard: 11,
outcome: BatchOutcome::Failed {
cause: Arc::new(DatabaseError::BatchFailed(Box::new(failure))),
},
}],
None,
)?;
let encoded = EncodedReceipt::project(&nested, None)?;
let cap = u32::try_from(minimum_cap()?)?;
let frames = encoded.frames("shard-cause", cap)?;
assert!(frames.len() > 2);
let mut assembler = Assembler::new("shard-cause", cap)?;
let mut complete = None;
for frame in frames {
assert!(frame.len() <= usize::try_from(cap)?);
if let Some(receipt) = assembler.accept(&frame)? {
assert!(complete.is_none());
complete = Some(receipt);
}
}
let complete = complete.ok_or("assembler never completed the receipt")?;
let wire = complete.receipt().ok_or("packet has no receipt")?;
let first = wire.first_error.as_ref().ok_or("first error missing")?;
assert_eq!(first.kind, DatabaseErrorKind::BatchShardFailed);
let shard = first.shard.as_ref().ok_or("shard cause missing")?;
assert_eq!(shard.kind, ShardErrorKind::Spawn);
assert_eq!(
shard.message,
"shard spawn failed: ".to_owned() + &"complete original shard cause".repeat(4096)
);
assert_eq!(Arc::strong_count(&cause), 3);
assert_eq!(wire, &WireReceipt::project(&nested)?);
Ok(())
}
#[test]
fn real_committed_callback_panic_roundtrips_and_reopens() -> TestResult {
use super::receipt_fragments::{Assembler, EncodedReceipt, minimum_cap};
use haematite::{Database, DatabaseConfig, NodeCacheBudget, Retention, RuntimeOptions};
use std::time::Duration;
let directory = tempfile::tempdir()?;
let result = (|| -> TestResult {
let path = directory.path().join("database");
let fixture_runtime = RuntimeOptions {
long_wait_interval: Duration::from_secs(1),
reclaim_start: haematite::ReclaimStart::AtOpen,
};
let database = Database::create(
DatabaseConfig {
data_dir: path.clone(),
shard_count: 1,
distributed: None,
executor_threads: Some(1),
node_cache_budget: Some(NodeCacheBudget::Unlimited),
retention: Retention::KeepAll,
},
fixture_runtime,
)?;
database.put(b"committed".to_vec(), b"survives".to_vec())?;
let marker = Arc::new(());
let callback_marker = Arc::clone(&marker);
let subscription = database.subscribe_root_advance(move |_| {
std::panic::resume_unwind(Box::new(Arc::clone(&callback_marker)));
});
let Err(error) = database.commit() else {
return Err("commit succeeded past a panicking root callback".into());
};
let DatabaseError::BatchFailed(failure) = &error else {
let detail = format!("real callback failure did not retain its batch receipt: {error}");
return Err(detail.into());
};
let receipt = failure.receipt();
let BatchOutcome::Committed {
root,
callback_failure: Some(cause),
} = &receipt.outcomes()[0].outcome
else {
return Err("callback panic lost the actual committed root".into());
};
let DatabaseError::RootCallbackPanicked { cause: panic } = cause.as_ref() else {
return Err("original callback panic cause was replaced".into());
};
assert_eq!(panic.event().new_root, *root);
let retained_owners = Arc::strong_count(&marker);
let encoded = EncodedReceipt::project(receipt, None)?;
let cap = u32::try_from(minimum_cap()?)?;
let mut assembler = Assembler::new("callback", cap)?;
let mut complete = None;
for frame in encoded.frames("callback", cap)? {
if let Some(receipt) = assembler.accept(&frame)? {
assert!(complete.is_none());
complete = Some(receipt);
}
}
let complete = complete.ok_or("assembler never completed the receipt")?;
assert_eq!(complete.bytes(), encoded.bytes());
let wire = complete.receipt().ok_or("packet has no receipt")?;
let WireOutcome::Committed {
root: wire_root,
callback_failure: Some(wire_cause),
} = &wire.outcomes[0].outcome
else {
return Err("wire receipt lost committed callback failure".into());
};
assert_eq!(wire_root, root.as_bytes());
assert_eq!(wire_cause.kind, DatabaseErrorKind::RootCallbackPanicked);
let metadata = wire_cause.callback.as_ref().ok_or("no callback record")?;
assert_eq!(metadata.pid, panic.identity().pid);
assert_eq!(
metadata.subscriber_index,
u64::try_from(panic.subscriber_index())?
);
assert_eq!(metadata.event.new_root, *root.as_bytes());
assert_eq!(Arc::strong_count(&marker), retained_owners);
drop(subscription);
drop(database);
let reopened = Database::open(path, fixture_runtime)?;
assert_eq!(reopened.get(b"committed")?, Some(b"survives".to_vec()));
drop(reopened);
Ok(())
})();
let cleanup = directory.close();
assert!(
result.is_ok() && cleanup.is_ok(),
"operation: {result:?}; cleanup: {cleanup:?}"
);
Ok(())
}