use std::collections::BTreeMap;
use std::fs::{File, OpenOptions};
use std::io::{self, Read, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Mutex;
use anyhow::{Context, Result};
use flatbuffers::FlatBufferBuilder;
use crate::generated::{
iam_wal_v1_generated::nexus::wal as fb_iam,
migration_wal_v1_generated::nexus::wal as fb_migration,
multipart_wal_v1_generated::nexus::wal as fb_multipart,
object_wal_v1_generated::nexus::wal as fb_object,
webhook_event_wal_v1_generated::nexus::wal as fb_webhook_event,
webhook_wal_v1_generated::nexus::wal as fb_webhook,
};
const FRAME_HEADER_BYTES: usize = 9;
const RECORD_KIND_OBJECT: u8 = 1;
const RECORD_KIND_MULTIPART: u8 = 2;
const RECORD_KIND_IAM: u8 = 3;
const RECORD_KIND_WEBHOOK: u8 = 4;
const RECORD_KIND_WEBHOOK_EVENT: u8 = 5;
const RECORD_KIND_MIGRATION: u8 = 6;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DevicePointer {
pub device_id: u16,
pub offset: u64,
pub len: u32,
pub checksum_crc32c: u32,
pub nonce_id: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WalEntryV1 {
pub seq: u64,
pub op: String,
pub bucket: String,
pub key: String,
pub etag: Option<String>,
pub pointer_or_manifest: Option<DevicePointer>,
pub ts_unix_ms: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MultipartPart {
pub part_number: u32,
pub etag: String,
pub pointer: DevicePointer,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MultipartManifest {
pub upload_id: String,
pub bucket: String,
pub key: String,
pub parts: Vec<MultipartPart>,
pub final_etag: String,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MultipartWalEntryV1 {
pub seq: u64,
pub op: String,
pub upload_id: String,
pub bucket: String,
pub key: String,
pub final_etag: Option<String>,
pub parts: Vec<MultipartPart>,
pub ts_unix_ms: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct IamWalEntryV1 {
pub seq: u64,
pub op: String,
pub target: String,
pub payload: String,
pub ts_unix_ms: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WebhookWalEntryV1 {
pub seq: u64,
pub op: String,
pub bucket: String,
pub rule_id: String,
pub rule_json: String,
pub ts_unix_ms: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WebhookEventWalEntryV1 {
pub seq: u64,
pub op: String,
pub event_id: String,
pub rule_id: String,
pub bucket: String,
pub key: String,
pub payload_json: String,
pub ts_unix_ms: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MigrationWalEntryV1 {
pub seq: u64,
pub op: String,
pub bucket: String,
pub key: String,
pub payload: String,
pub ts_unix_ms: u64,
}
#[derive(Debug, Clone)]
pub struct WalReplayConfig {
pub wal_path: PathBuf,
pub dry_run: bool,
}
#[derive(Debug, Clone)]
pub struct WalReplayStats {
pub wal_path: PathBuf,
pub dry_run: bool,
pub object_entries: u64,
pub multipart_entries: u64,
pub object_puts: u64,
pub object_deletes: u64,
pub multipart_creates: u64,
pub multipart_parts: u64,
pub multipart_completes: u64,
pub multipart_aborts: u64,
pub iam_entries: u64,
pub webhook_entries: u64,
pub webhook_event_entries: u64,
pub migration_entries: u64,
pub last_seq: u64,
}
impl WalReplayStats {
pub fn to_json(&self) -> String {
format!(
"{{\"module\":\"WAL_REPLAY\",\"wal_path\":\"{}\",\"dry_run\":{},\"object_entries\":{},\"multipart_entries\":{},\"object_puts\":{},\"object_deletes\":{},\"multipart_creates\":{},\"multipart_parts\":{},\"multipart_completes\":{},\"multipart_aborts\":{},\"iam_entries\":{},\"webhook_entries\":{},\"webhook_event_entries\":{},\"migration_entries\":{},\"last_seq\":{}}}",
self.wal_path.display(),
self.dry_run,
self.object_entries,
self.multipart_entries,
self.object_puts,
self.object_deletes,
self.multipart_creates,
self.multipart_parts,
self.multipart_completes,
self.multipart_aborts,
self.iam_entries,
self.webhook_entries,
self.webhook_event_entries,
self.migration_entries,
self.last_seq,
)
}
}
#[derive(Debug, Clone)]
pub struct WalReplayState {
pub stats: WalReplayStats,
pub object_index: BTreeMap<String, WalEntryV1>,
pub multipart_uploads: BTreeMap<String, MultipartWalEntryV1>,
pub iam_log: Vec<IamWalEntryV1>,
pub webhook_log: Vec<WebhookWalEntryV1>,
pub webhook_event_log: Vec<WebhookEventWalEntryV1>,
pub migration_log: Vec<MigrationWalEntryV1>,
}
#[derive(Debug)]
pub struct WalWriter {
wal_path: PathBuf,
file: Mutex<File>,
next_seq: AtomicU64,
}
impl WalWriter {
pub fn open<P: AsRef<Path>>(wal_path: P) -> Result<Self> {
let wal_path = wal_path.as_ref().to_path_buf();
if let Some(parent) = wal_path.parent() {
if !parent.as_os_str().is_empty() {
std::fs::create_dir_all(parent)
.with_context(|| format!("failed creating WAL parent {}", parent.display()))?;
}
}
let file = OpenOptions::new()
.create(true)
.append(true)
.read(true)
.open(&wal_path)
.with_context(|| format!("failed opening WAL {}", wal_path.display()))?;
let seed_seq = detect_next_seq(&wal_path).unwrap_or(1);
Ok(Self {
wal_path,
file: Mutex::new(file),
next_seq: AtomicU64::new(seed_seq),
})
}
pub fn wal_path(&self) -> &Path {
&self.wal_path
}
pub fn next_seq(&self) -> u64 {
self.next_seq.fetch_add(1, Ordering::AcqRel)
}
pub fn append_object_entry(&self, entry: &WalEntryV1) -> Result<()> {
let mut fbb = FlatBufferBuilder::with_capacity(512);
let op = fbb.create_string(entry.op.as_str());
let bucket = fbb.create_string(entry.bucket.as_str());
let key = fbb.create_string(entry.key.as_str());
let etag = entry.etag.as_ref().map(|v| fbb.create_string(v.as_str()));
let pointer = entry.pointer_or_manifest.as_ref().map(|p| {
fb_object::DevicePointer::create(
&mut fbb,
&fb_object::DevicePointerArgs {
device_id: p.device_id,
offset: p.offset,
len: p.len,
checksum_crc32c: p.checksum_crc32c,
nonce_id: p.nonce_id,
},
)
});
let record = fb_object::ObjectRecordV1::create(
&mut fbb,
&fb_object::ObjectRecordV1Args {
seq: entry.seq,
op: Some(op),
bucket: Some(bucket),
key: Some(key),
etag,
ts_unix_ms: entry.ts_unix_ms,
pointer,
},
);
fb_object::finish_size_prefixed_object_record_v1_buffer(&mut fbb, record);
self.append_frame(RECORD_KIND_OBJECT, fbb.finished_data())
}
pub fn append_multipart_entry(&self, entry: &MultipartWalEntryV1) -> Result<()> {
let mut fbb = FlatBufferBuilder::with_capacity(1024);
let op = fbb.create_string(entry.op.as_str());
let bucket = fbb.create_string(entry.bucket.as_str());
let key = fbb.create_string(entry.key.as_str());
let upload_id = fbb.create_string(entry.upload_id.as_str());
let final_etag = entry
.final_etag
.as_ref()
.map(|v| fbb.create_string(v.as_str()));
let mut part_offsets = Vec::with_capacity(entry.parts.len());
for part in &entry.parts {
let etag = fbb.create_string(part.etag.as_str());
let pointer = fb_multipart::DevicePointer::create(
&mut fbb,
&fb_multipart::DevicePointerArgs {
device_id: part.pointer.device_id,
offset: part.pointer.offset,
len: part.pointer.len,
checksum_crc32c: part.pointer.checksum_crc32c,
nonce_id: part.pointer.nonce_id,
},
);
let part_entry = fb_multipart::MultipartPartPointer::create(
&mut fbb,
&fb_multipart::MultipartPartPointerArgs {
part_number: part.part_number,
etag: Some(etag),
pointer: Some(pointer),
},
);
part_offsets.push(part_entry);
}
let parts = if part_offsets.is_empty() {
None
} else {
Some(fbb.create_vector(&part_offsets))
};
let record = fb_multipart::MultipartRecordV1::create(
&mut fbb,
&fb_multipart::MultipartRecordV1Args {
seq: entry.seq,
op: Some(op),
bucket: Some(bucket),
key: Some(key),
upload_id: Some(upload_id),
final_etag,
ts_unix_ms: entry.ts_unix_ms,
parts,
},
);
fb_multipart::finish_size_prefixed_multipart_record_v1_buffer(&mut fbb, record);
self.append_frame(RECORD_KIND_MULTIPART, fbb.finished_data())
}
pub fn append_iam_entry(&self, entry: &IamWalEntryV1) -> Result<()> {
let mut fbb = FlatBufferBuilder::with_capacity(512);
let op = fbb.create_string(entry.op.as_str());
let target = fbb.create_string(entry.target.as_str());
let payload = fbb.create_string(entry.payload.as_str());
let record = fb_iam::IamRecordV1::create(
&mut fbb,
&fb_iam::IamRecordV1Args {
seq: entry.seq,
op: Some(op),
target: Some(target),
payload: Some(payload),
ts_unix_ms: entry.ts_unix_ms,
},
);
fb_iam::finish_size_prefixed_iam_record_v1_buffer(&mut fbb, record);
self.append_frame(RECORD_KIND_IAM, fbb.finished_data())
}
pub fn append_webhook_entry(&self, entry: &WebhookWalEntryV1) -> Result<()> {
let mut fbb = FlatBufferBuilder::with_capacity(512);
let op = fbb.create_string(entry.op.as_str());
let bucket = fbb.create_string(entry.bucket.as_str());
let rule_id = fbb.create_string(entry.rule_id.as_str());
let rule_json = fbb.create_string(entry.rule_json.as_str());
let record = fb_webhook::WebhookRecordV1::create(
&mut fbb,
&fb_webhook::WebhookRecordV1Args {
seq: entry.seq,
op: Some(op),
bucket: Some(bucket),
rule_id: Some(rule_id),
rule_json: Some(rule_json),
ts_unix_ms: entry.ts_unix_ms,
},
);
fb_webhook::finish_size_prefixed_webhook_record_v1_buffer(&mut fbb, record);
self.append_frame(RECORD_KIND_WEBHOOK, fbb.finished_data())
}
pub fn append_webhook_event_entry(&self, entry: &WebhookEventWalEntryV1) -> Result<()> {
let mut fbb = FlatBufferBuilder::with_capacity(512);
let op = fbb.create_string(entry.op.as_str());
let event_id = fbb.create_string(entry.event_id.as_str());
let rule_id = fbb.create_string(entry.rule_id.as_str());
let bucket = fbb.create_string(entry.bucket.as_str());
let key = fbb.create_string(entry.key.as_str());
let payload_json = fbb.create_string(entry.payload_json.as_str());
let record = fb_webhook_event::WebhookEventRecordV1::create(
&mut fbb,
&fb_webhook_event::WebhookEventRecordV1Args {
seq: entry.seq,
op: Some(op),
event_id: Some(event_id),
rule_id: Some(rule_id),
bucket: Some(bucket),
key: Some(key),
payload_json: Some(payload_json),
ts_unix_ms: entry.ts_unix_ms,
},
);
fb_webhook_event::finish_size_prefixed_webhook_event_record_v1_buffer(&mut fbb, record);
self.append_frame(RECORD_KIND_WEBHOOK_EVENT, fbb.finished_data())
}
pub fn append_migration_entry(&self, entry: &MigrationWalEntryV1) -> Result<()> {
let mut fbb = FlatBufferBuilder::with_capacity(512);
let op = fbb.create_string(entry.op.as_str());
let bucket = fbb.create_string(entry.bucket.as_str());
let key = fbb.create_string(entry.key.as_str());
let payload = fbb.create_string(entry.payload.as_str());
let record = fb_migration::MigrationRecordV1::create(
&mut fbb,
&fb_migration::MigrationRecordV1Args {
seq: entry.seq,
op: Some(op),
bucket: Some(bucket),
key: Some(key),
payload: Some(payload),
ts_unix_ms: entry.ts_unix_ms,
},
);
fb_migration::finish_size_prefixed_migration_record_v1_buffer(&mut fbb, record);
self.append_frame(RECORD_KIND_MIGRATION, fbb.finished_data())
}
fn append_frame(&self, kind: u8, payload: &[u8]) -> Result<()> {
let payload_len: u32 = payload
.len()
.try_into()
.context("WAL payload too large to frame")?;
let payload_crc = crc32c::crc32c(payload);
let mut header = [0_u8; FRAME_HEADER_BYTES];
header[0] = kind;
header[1..5].copy_from_slice(&payload_len.to_le_bytes());
header[5..9].copy_from_slice(&payload_crc.to_le_bytes());
let mut file = self
.file
.lock()
.map_err(|_| anyhow::anyhow!("WAL file mutex poisoned"))?;
file.write_all(&header)
.context("failed writing WAL frame header")?;
file.write_all(payload)
.context("failed writing WAL frame payload")?;
file.sync_data().context("failed syncing WAL append")?;
Ok(())
}
}
pub fn replay_wal(config: WalReplayConfig) -> Result<WalReplayStats> {
let state = replay_wal_state(config)?;
Ok(state.stats)
}
pub fn replay_wal_state(config: WalReplayConfig) -> Result<WalReplayState> {
if !config.wal_path.exists() {
return Ok(WalReplayState {
stats: WalReplayStats {
wal_path: config.wal_path,
dry_run: config.dry_run,
object_entries: 0,
multipart_entries: 0,
object_puts: 0,
object_deletes: 0,
multipart_creates: 0,
multipart_parts: 0,
multipart_completes: 0,
multipart_aborts: 0,
iam_entries: 0,
webhook_entries: 0,
webhook_event_entries: 0,
migration_entries: 0,
last_seq: 0,
},
object_index: BTreeMap::new(),
multipart_uploads: BTreeMap::new(),
iam_log: Vec::new(),
webhook_log: Vec::new(),
webhook_event_log: Vec::new(),
migration_log: Vec::new(),
});
}
let mut file = File::open(&config.wal_path)
.with_context(|| format!("failed opening WAL {}", config.wal_path.display()))?;
let mut object_index = BTreeMap::<String, WalEntryV1>::new();
let mut multipart_uploads = BTreeMap::<String, MultipartWalEntryV1>::new();
let mut stats = WalReplayStats {
wal_path: config.wal_path.clone(),
dry_run: config.dry_run,
object_entries: 0,
multipart_entries: 0,
object_puts: 0,
object_deletes: 0,
multipart_creates: 0,
multipart_parts: 0,
multipart_completes: 0,
multipart_aborts: 0,
iam_entries: 0,
webhook_entries: 0,
webhook_event_entries: 0,
migration_entries: 0,
last_seq: 0,
};
let mut iam_log = Vec::<IamWalEntryV1>::new();
let mut webhook_log = Vec::<WebhookWalEntryV1>::new();
let mut webhook_event_log = Vec::<WebhookEventWalEntryV1>::new();
let mut migration_log = Vec::<MigrationWalEntryV1>::new();
loop {
let mut header = [0_u8; FRAME_HEADER_BYTES];
match file.read_exact(&mut header) {
Ok(()) => {}
Err(err) if err.kind() == io::ErrorKind::UnexpectedEof => break,
Err(err) => return Err(err).context("failed reading WAL frame header"),
}
let kind = header[0];
let payload_len = u32::from_le_bytes(header[1..5].try_into().expect("header len"));
let payload_crc = u32::from_le_bytes(header[5..9].try_into().expect("header crc"));
if kind == 0 && payload_len == 0 && payload_crc == 0 {
break;
}
let mut payload = vec![0_u8; payload_len as usize];
file.read_exact(&mut payload)
.context("failed reading WAL frame payload")?;
let actual_crc = crc32c::crc32c(&payload);
if actual_crc != payload_crc {
anyhow::bail!(
"WAL CRC mismatch at seq {}: expected {}, got {}",
stats.last_seq,
payload_crc,
actual_crc
);
}
match kind {
RECORD_KIND_OBJECT => {
stats.object_entries += 1;
let record = fb_object::size_prefixed_root_as_object_record_v1(&payload)
.map_err(|err| anyhow::anyhow!("invalid object WAL frame: {err}"))?;
let entry = decode_object_record(record)?;
enforce_monotonic_seq(stats.last_seq, entry.seq)?;
stats.last_seq = entry.seq;
match entry.op.as_str() {
"put" => {
stats.object_puts += 1;
if !config.dry_run {
object_index.insert(format!("{}/{}", entry.bucket, entry.key), entry);
}
}
"delete" => {
stats.object_deletes += 1;
if !config.dry_run {
object_index.remove(&format!("{}/{}", entry.bucket, entry.key));
}
}
other => anyhow::bail!("unsupported object WAL op: {other}"),
}
}
RECORD_KIND_MULTIPART => {
stats.multipart_entries += 1;
let record = fb_multipart::size_prefixed_root_as_multipart_record_v1(&payload)
.map_err(|err| anyhow::anyhow!("invalid multipart WAL frame: {err}"))?;
let entry = decode_multipart_record(record)?;
enforce_monotonic_seq(stats.last_seq, entry.seq)?;
stats.last_seq = entry.seq;
match entry.op.as_str() {
"create" => {
stats.multipart_creates += 1;
if !config.dry_run {
multipart_uploads.insert(entry.upload_id.clone(), entry);
}
}
"upload_part" => {
stats.multipart_parts += 1;
if !config.dry_run {
let state = multipart_uploads
.entry(entry.upload_id.clone())
.or_insert_with(|| MultipartWalEntryV1 {
seq: entry.seq,
op: "create".to_string(),
upload_id: entry.upload_id.clone(),
bucket: entry.bucket.clone(),
key: entry.key.clone(),
final_etag: None,
parts: Vec::new(),
ts_unix_ms: entry.ts_unix_ms,
});
for incoming in entry.parts {
if let Some(existing) = state
.parts
.iter_mut()
.find(|part| part.part_number == incoming.part_number)
{
*existing = incoming;
} else {
state.parts.push(incoming);
}
}
state.parts.sort_by_key(|part| part.part_number);
state.seq = entry.seq;
state.ts_unix_ms = entry.ts_unix_ms;
}
}
"complete" => {
stats.multipart_completes += 1;
if !config.dry_run {
multipart_uploads.insert(entry.upload_id.clone(), entry);
}
}
"abort" => {
stats.multipart_aborts += 1;
if !config.dry_run {
multipart_uploads.remove(&entry.upload_id);
}
}
other => anyhow::bail!("unsupported multipart WAL op: {other}"),
}
}
RECORD_KIND_IAM => {
stats.iam_entries += 1;
let record = fb_iam::size_prefixed_root_as_iam_record_v1(&payload)
.map_err(|err| anyhow::anyhow!("invalid IAM WAL frame: {err}"))?;
let entry = decode_iam_record(record)?;
enforce_monotonic_seq(stats.last_seq, entry.seq)?;
stats.last_seq = entry.seq;
if !config.dry_run {
iam_log.push(entry);
}
}
RECORD_KIND_WEBHOOK => {
stats.webhook_entries += 1;
let record = fb_webhook::size_prefixed_root_as_webhook_record_v1(&payload)
.map_err(|err| anyhow::anyhow!("invalid webhook WAL frame: {err}"))?;
let entry = decode_webhook_record(record)?;
enforce_monotonic_seq(stats.last_seq, entry.seq)?;
stats.last_seq = entry.seq;
if !config.dry_run {
webhook_log.push(entry);
}
}
RECORD_KIND_WEBHOOK_EVENT => {
stats.webhook_event_entries += 1;
let record =
fb_webhook_event::size_prefixed_root_as_webhook_event_record_v1(&payload)
.map_err(|err| {
anyhow::anyhow!("invalid webhook event WAL frame: {err}")
})?;
let entry = decode_webhook_event_record(record)?;
enforce_monotonic_seq(stats.last_seq, entry.seq)?;
stats.last_seq = entry.seq;
if !config.dry_run {
webhook_event_log.push(entry);
}
}
RECORD_KIND_MIGRATION => {
stats.migration_entries += 1;
let record = fb_migration::size_prefixed_root_as_migration_record_v1(&payload)
.map_err(|err| anyhow::anyhow!("invalid migration WAL frame: {err}"))?;
let entry = decode_migration_record(record)?;
enforce_monotonic_seq(stats.last_seq, entry.seq)?;
stats.last_seq = entry.seq;
if !config.dry_run {
migration_log.push(entry);
}
}
other => anyhow::bail!("unknown WAL record kind: {other}"),
}
}
Ok(WalReplayState {
stats,
object_index,
multipart_uploads,
iam_log,
webhook_log,
webhook_event_log,
migration_log,
})
}
fn decode_object_record(record: fb_object::ObjectRecordV1<'_>) -> Result<WalEntryV1> {
let op = record
.op()
.context("object WAL record missing op")?
.to_string();
let bucket = record
.bucket()
.context("object WAL record missing bucket")?
.to_string();
let key = record
.key()
.context("object WAL record missing key")?
.to_string();
let pointer_or_manifest = record.pointer().map(|ptr| DevicePointer {
device_id: ptr.device_id(),
offset: ptr.offset(),
len: ptr.len(),
checksum_crc32c: ptr.checksum_crc32c(),
nonce_id: ptr.nonce_id(),
});
Ok(WalEntryV1 {
seq: record.seq(),
op,
bucket,
key,
etag: record.etag().map(ToString::to_string),
pointer_or_manifest,
ts_unix_ms: record.ts_unix_ms(),
})
}
fn decode_multipart_record(
record: fb_multipart::MultipartRecordV1<'_>,
) -> Result<MultipartWalEntryV1> {
let op = record
.op()
.context("multipart WAL record missing op")?
.to_string();
let bucket = record
.bucket()
.context("multipart WAL record missing bucket")?
.to_string();
let key = record
.key()
.context("multipart WAL record missing key")?
.to_string();
let upload_id = record
.upload_id()
.context("multipart WAL record missing upload_id")?
.to_string();
let mut parts = Vec::new();
if let Some(list) = record.parts() {
for part in list {
let pointer = part
.pointer()
.context("multipart WAL part missing pointer")?;
parts.push(MultipartPart {
part_number: part.part_number(),
etag: part.etag().unwrap_or_default().to_string(),
pointer: DevicePointer {
device_id: pointer.device_id(),
offset: pointer.offset(),
len: pointer.len(),
checksum_crc32c: pointer.checksum_crc32c(),
nonce_id: pointer.nonce_id(),
},
});
}
}
Ok(MultipartWalEntryV1 {
seq: record.seq(),
op,
upload_id,
bucket,
key,
final_etag: record.final_etag().map(ToString::to_string),
parts,
ts_unix_ms: record.ts_unix_ms(),
})
}
fn decode_iam_record(record: fb_iam::IamRecordV1<'_>) -> Result<IamWalEntryV1> {
Ok(IamWalEntryV1 {
seq: record.seq(),
op: record
.op()
.context("iam WAL record missing op")?
.to_string(),
target: record
.target()
.context("iam WAL record missing target")?
.to_string(),
payload: record
.payload()
.context("iam WAL record missing payload")?
.to_string(),
ts_unix_ms: record.ts_unix_ms(),
})
}
fn decode_webhook_record(record: fb_webhook::WebhookRecordV1<'_>) -> Result<WebhookWalEntryV1> {
Ok(WebhookWalEntryV1 {
seq: record.seq(),
op: record
.op()
.context("webhook WAL record missing op")?
.to_string(),
bucket: record
.bucket()
.context("webhook WAL record missing bucket")?
.to_string(),
rule_id: record
.rule_id()
.context("webhook WAL record missing rule_id")?
.to_string(),
rule_json: record
.rule_json()
.context("webhook WAL record missing rule_json")?
.to_string(),
ts_unix_ms: record.ts_unix_ms(),
})
}
fn decode_webhook_event_record(
record: fb_webhook_event::WebhookEventRecordV1<'_>,
) -> Result<WebhookEventWalEntryV1> {
Ok(WebhookEventWalEntryV1 {
seq: record.seq(),
op: record
.op()
.context("webhook event WAL record missing op")?
.to_string(),
event_id: record
.event_id()
.context("webhook event WAL record missing event_id")?
.to_string(),
rule_id: record
.rule_id()
.context("webhook event WAL record missing rule_id")?
.to_string(),
bucket: record
.bucket()
.context("webhook event WAL record missing bucket")?
.to_string(),
key: record
.key()
.context("webhook event WAL record missing key")?
.to_string(),
payload_json: record
.payload_json()
.context("webhook event WAL record missing payload_json")?
.to_string(),
ts_unix_ms: record.ts_unix_ms(),
})
}
fn decode_migration_record(record: fb_migration::MigrationRecordV1<'_>) -> Result<MigrationWalEntryV1> {
Ok(MigrationWalEntryV1 {
seq: record.seq(),
op: record
.op()
.context("migration WAL record missing op")?
.to_string(),
bucket: record
.bucket()
.context("migration WAL record missing bucket")?
.to_string(),
key: record
.key()
.context("migration WAL record missing key")?
.to_string(),
payload: record
.payload()
.context("migration WAL record missing payload")?
.to_string(),
ts_unix_ms: record.ts_unix_ms(),
})
}
fn detect_next_seq(path: &Path) -> Option<u64> {
let state = replay_wal_state(WalReplayConfig {
wal_path: path.to_path_buf(),
dry_run: true,
})
.ok()?;
Some(state.stats.last_seq.saturating_add(1).max(1))
}
fn enforce_monotonic_seq(previous: u64, next: u64) -> Result<()> {
if next <= previous {
anyhow::bail!(
"WAL sequence monotonicity violation: previous={}, next={}",
previous,
next
);
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn wal_round_trip_object_and_multipart() {
let wal_path = unique_temp_path("wal-roundtrip");
let writer = WalWriter::open(&wal_path).expect("open WAL writer");
let seq1 = writer.next_seq();
writer
.append_object_entry(&WalEntryV1 {
seq: seq1,
op: "put".to_string(),
bucket: "b".to_string(),
key: "k".to_string(),
etag: Some("etag-1".to_string()),
pointer_or_manifest: Some(DevicePointer {
device_id: 1,
offset: 8192,
len: 4096,
checksum_crc32c: 123,
nonce_id: 77,
}),
ts_unix_ms: 1,
})
.expect("append object entry");
let seq2 = writer.next_seq();
writer
.append_multipart_entry(&MultipartWalEntryV1 {
seq: seq2,
op: "create".to_string(),
upload_id: "u1".to_string(),
bucket: "b".to_string(),
key: "k2".to_string(),
final_etag: None,
parts: Vec::new(),
ts_unix_ms: 2,
})
.expect("append multipart entry");
let replay = replay_wal_state(WalReplayConfig {
wal_path: wal_path.clone(),
dry_run: false,
})
.expect("replay should succeed");
assert_eq!(replay.stats.object_entries, 1);
assert_eq!(replay.stats.multipart_entries, 1);
assert_eq!(replay.stats.last_seq, seq2);
assert!(replay.object_index.contains_key("b/k"));
assert!(replay.multipart_uploads.contains_key("u1"));
let _ = std::fs::remove_file(wal_path);
}
#[test]
fn wal_replay_rejects_non_monotonic_sequence() {
let wal_path = unique_temp_path("wal-bad-seq");
let writer = WalWriter::open(&wal_path).expect("open WAL writer");
writer
.append_object_entry(&WalEntryV1 {
seq: 3,
op: "put".to_string(),
bucket: "b".to_string(),
key: "k1".to_string(),
etag: None,
pointer_or_manifest: None,
ts_unix_ms: 1,
})
.expect("append seq3");
writer
.append_object_entry(&WalEntryV1 {
seq: 2,
op: "put".to_string(),
bucket: "b".to_string(),
key: "k2".to_string(),
etag: None,
pointer_or_manifest: None,
ts_unix_ms: 2,
})
.expect("append seq2");
let err = replay_wal(WalReplayConfig {
wal_path: wal_path.clone(),
dry_run: false,
})
.expect_err("expected monotonicity violation");
assert!(err.to_string().contains("monotonicity"));
let _ = std::fs::remove_file(wal_path);
}
fn unique_temp_path(prefix: &str) -> PathBuf {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.expect("valid system time")
.as_nanos();
std::env::temp_dir().join(format!("{prefix}-{}-{now}.wal", std::process::id()))
}
}