diff --git a/changelog.d/sh-reorg-unlink.md b/changelog.d/sh-reorg-unlink.md new file mode 100644 index 0000000000..9789222f6f --- /dev/null +++ b/changelog.d/sh-reorg-unlink.md @@ -0,0 +1,6 @@ +Fixed + +- **A one-block reorg no longer rewrites a large scripthash once per output.** + Disconnect drops that block's creates from the tail of the key. The previous + pages stay readable until the new tail is the head. A failed undo clear puts + that tail back, and a full disk cannot tear the index. diff --git a/crates/rbitcoin-query/src/connect.rs b/crates/rbitcoin-query/src/connect.rs index be3f803b28..d020bb07e5 100644 --- a/crates/rbitcoin-query/src/connect.rs +++ b/crates/rbitcoin-query/src/connect.rs @@ -854,7 +854,8 @@ impl Query { /// tip is still high (permanent if kill). Mid-kill after tip shrink leaves /// strong **not on the new fence** → `repair_class_c_above_tip` heals. /// - /// Every successful tip shrink logs [`format_disconnect_tip_line`] at **warn**. + /// Logs [`format_disconnect_tip_line`] at **warn** before the scripthash + /// unlink and again after the tip shrink commits. pub fn disconnect_tip(&self) -> Result<(), QueryError> { let _commit = self.lock_tip_commit(); self.disconnect_tip_body(true) @@ -899,15 +900,22 @@ impl Query { .unwrap_or([0u8; 32]); let tx_fks = self.block_tx_fks(height)?; - let mut touched_sh: Vec<[u8; 32]> = Vec::new(); + log_disconnect_tip(height.0, &hash, tx_fks.len()); + let mut by_sh: HashMap<[u8; 32], Vec> = HashMap::new(); for &tx_fk in &tx_fks { let outputs = self.tx_output_run_class_a(tx_fk)?; - for (i, o) in outputs.iter().enumerate() { - let sh = script_hash(&o.script); - let _ = self.store.scripthash.unlink_create(&sh, tx_fk, i as u32)?; - touched_sh.push(sh); + for o in &outputs { + by_sh.entry(script_hash(&o.script)).or_default().push(tx_fk); } } + let mut groups: Vec<([u8; 32], Vec)> = Vec::with_capacity(by_sh.len()); + for (sh, mut fks) in by_sh { + fks.sort_by_key(|fk| fk.0); + fks.dedup(); + groups.push((sh, fks)); + } + self.store.scripthash.unlink_creates_batch(&groups)?; + let touched_sh: Vec<[u8; 32]> = groups.iter().map(|(sh, _)| *sh).collect(); if !touched_sh.is_empty() { let mut heads = self.sh.heads.lock().unwrap(); for sh in touched_sh { diff --git a/crates/rbitcoin-store/src/scripthash.rs b/crates/rbitcoin-store/src/scripthash.rs index 28467236eb..41fbce1995 100644 --- a/crates/rbitcoin-store/src/scripthash.rs +++ b/crates/rbitcoin-store/src/scripthash.rs @@ -19,9 +19,9 @@ use crate::scripthash_mphf::{mix_key16, MphfHead}; use crate::scripthash_overflow::wipe_legacy_fullsize_overflow; use crate::scripthash_pages::{ sh_page_as_array, sh_page_chunk_ranges, sh_page_decode_slice_into, sh_page_extent, - sh_page_first_off, sh_page_init_empty, sh_page_is_last, sh_page_last_fk, sh_page_next, - sh_page_pack_extent_last_fks, sh_page_pack_fks, sh_page_set_last, sh_page_set_next, - sh_page_try_append, SH_PAGE_SIZE, SH_PAGE_STREAM_MAX, + sh_page_first_off, sh_page_is_last, sh_page_last_fk, sh_page_next, + sh_page_pack_extent_last_fks, sh_page_pack_fks, sh_page_set_next, sh_page_try_append, + SH_PAGE_SIZE, SH_PAGE_STREAM_MAX, }; use crate::scripthash_slabs::{ decode_slab_payload_into, encode_slab_payload_into, slab_class_for_n_fks_with_slack, @@ -35,6 +35,9 @@ use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Mutex, RwLock}; +#[path = "scripthash_unlink.rs"] +mod unlink; + /// Durable cold-materialize resume marker (next to `scripthash.head`). pub const COLD_PROGRESS_NAME: &str = "scripthash.cold_progress"; const COLD_PROGRESS_MAGIC: &[u8; 8] = b"SHCOLDP1"; @@ -619,7 +622,11 @@ fn paged_first_from_last(body: &TableFile, last_page: u64) -> Result { let mut page = [0u8; SH_PAGE_SIZE]; body.read_at(*last_page, &mut page)?; + if sh_page_extent(&page)?.is_none() { + return Err(StoreError::Corrupt("scripthash extent last page not ver=2")); + } sh_page_last_fk(&page) } } @@ -1908,24 +1919,15 @@ impl ScriptHashTable { } } + /// Pack `live` into a new inline, slab, or extent. Does not free `old` + /// and does not publish the head. The caller writes the head, then frees. fn rewrite_entries_for_key( &self, body: &TableFile, alloc: &mut AllocState, - old: &ShHeadValue, live: &[Fk], ) -> Result { - let n = live.len() as u32; - let old_list = self.collect_entries_from(body, old)?; - let old_n = old_list.len() as u32; - if n > old_n { - alloc.live_count = alloc.live_count.saturating_add(u64::from(n - old_n)); - } else if n < old_n { - alloc.live_count = alloc.live_count.saturating_sub(u64::from(old_n - n)); - } - - if n == 0 { - self.free_if_paged(body, alloc, old)?; + if live.is_empty() { return Ok(ShHeadValue::Empty); } for w in live.windows(2) { @@ -1935,7 +1937,6 @@ impl ScriptHashTable { )); } } - self.free_if_paged(body, alloc, old)?; self.pack_entries(body, alloc, live, false) } @@ -1992,8 +1993,7 @@ impl ScriptHashTable { Ok(new_val) } ShHeadValue::Extent { last_page } => { - let first = paged_first_from_last(body, *last_page)?; - let last = self.append_fks_to_pages(body, alloc, first, *last_page, new_ents)?; + let last = self.append_fks_to_pages(body, alloc, *last_page, new_ents)?; Ok(ShHeadValue::extent(last)) } } @@ -2117,11 +2117,14 @@ impl ScriptHashTable { } /// Append `tail` FKs onto an existing chain ending at `last_page`. + /// + /// The last page is ver=2 with no successor. A zero creates counter is + /// recounted from the pages. A ver=1 tail is not produced here: the + /// extent check already rejected it. fn append_fks_to_pages( &self, body: &TableFile, alloc: &mut AllocState, - first_page: u64, last_page: u64, tail: &[Fk], ) -> Result { @@ -2131,21 +2134,18 @@ impl ScriptHashTable { let mut last = last_page; let mut page = [0u8; SH_PAGE_SIZE]; body.read_at(last, &mut page)?; - let mut creates = crate::scripthash_pages::sh_page_extent_creates(&page); - let mut extent = sh_page_extent(&page)?; - let chain_first = { - let f = if sh_page_is_last(&page)? { - sh_page_first_off(&page)? - } else { - 0 - }; - if f == 0 { - first_page - } else { - f - } + let Some((base, mut n)) = sh_page_extent(&page)? else { + return Err(StoreError::Corrupt("scripthash extent last page not ver=2")); }; - if creates == 0 && extent.is_some() { + // A low freelist tail and an inflated extent_n both sit below the + // stamped geometric end. The tail is the page marked last. + if !sh_page_is_last(&page)? { + return Err(StoreError::Corrupt( + "scripthash extent last page is not the extent tail", + )); + } + let mut creates = crate::scripthash_pages::sh_page_extent_creates(&page); + if creates == 0 { let existing = self.collect_entries_from(body, &ShHeadValue::extent(last_page))?; creates = u32::try_from(existing.len()).unwrap_or(u32::MAX); } @@ -2154,19 +2154,12 @@ impl ScriptHashTable { let new_off = self.alloc_page(body, alloc)?; sh_page_set_next(&mut page, new_off)?; body.write_at(last, &page)?; - if let Some((base, n)) = extent { - let glued = new_off == base.saturating_add(u64::from(n) * SH_PAGE_SIZE as u64); - let new_n = if glued { n.saturating_add(1) } else { n }; - sh_page_pack_extent_last_fks(&mut page, &[fk.0], base, new_n, 0)?; - if creates > 0 { - creates = creates.saturating_add(1); - crate::scripthash_pages::sh_page_set_extent_creates(&mut page, creates); - } - extent = Some((base, new_n)); - } else { - sh_page_init_empty(&mut page); - sh_page_set_last(&mut page, chain_first)?; - assert!(sh_page_try_append(&mut page, *fk)?); + let glued = new_off == base.saturating_add(u64::from(n) * SH_PAGE_SIZE as u64); + n = if glued { n.saturating_add(1) } else { n }; + sh_page_pack_extent_last_fks(&mut page, &[fk.0], base, n, 0)?; + if creates > 0 { + creates = creates.saturating_add(1); + crate::scripthash_pages::sh_page_set_extent_creates(&mut page, creates); } last = new_off; } else if creates > 0 { @@ -2274,34 +2267,6 @@ impl ScriptHashTable { push_free_head(body, &mut alloc.free_head, class, off) } - /// Unlink one create_tx_fk (disconnect tip). Caller should only remove the fk - /// when no remaining outputs of that tx still match this scripthash. - /// Swap-remove; demote paged→inline when ≤2 remain. - pub fn unlink_create( - &self, - scripthash: &[u8; 32], - create_tx_fk: Fk, - _vout: u32, - ) -> Result { - let Some((val, home)) = self.locate_head(scripthash)? else { - return Ok(false); - }; - let body = self.body_for(scripthash, KeyHome::from(home)); - let mut live = self.collect_entries_from(body, &val)?; - let Some(pos) = live.iter().position(|fk| *fk == create_tx_fk) else { - return Ok(false); - }; - live.remove(pos); - live.sort_by_key(|fk| fk.0); - let alloc_mu = self.alloc_for(scripthash, KeyHome::from(home)); - let mut alloc = alloc_mu.lock().unwrap(); - let new_val = self.rewrite_entries_for_key(body, &mut alloc, &val, &live)?; - write_alloc_header(body, &alloc)?; - drop(alloc); - self.unlink_write_home(scripthash, home, &new_val)?; - Ok(true) - } - fn unlink_write_home( &self, scripthash: &[u8; 32], @@ -3102,7 +3067,8 @@ impl<'a> ScriptHashBulkSession<'a> { } fn write_alloc_header(body: &TableFile, state: &AllocState) -> Result<(), StoreError> { - let mut buf = vec![0u8; SH_ALLOC_HEADER_LEN]; + // Stop at the freelist. Bytes after that hold an in-progress unlink undo. + let mut buf = vec![0u8; 24 + 8 * (SH_MAX_CLASS as usize + 1)]; buf[0..4].copy_from_slice(&SH_ALLOC_MAGIC); buf[4..6].copy_from_slice(&SH_ALLOC_VERSION.to_le_bytes()); buf[8..16].copy_from_slice(&state.live_count.to_le_bytes()); diff --git a/crates/rbitcoin-store/src/scripthash_head.rs b/crates/rbitcoin-store/src/scripthash_head.rs index a3818a9ab5..65e23a2ad1 100644 --- a/crates/rbitcoin-store/src/scripthash_head.rs +++ b/crates/rbitcoin-store/src/scripthash_head.rs @@ -263,6 +263,39 @@ impl ScriptHashHead { self.insert_many(&[(*full, value.clone())]) } + /// Byte offset of the pack8 word for an occupied key. + pub(crate) fn value_word_offset(&self, full: &[u8; 32]) -> Result, StoreError> { + let key = Self::to_key(full); + let slots = self.state.lock().unwrap().slots; + let mut cache = SlotPageCache::new(self, slots); + let mut slot = Self::hash_slot(&key, slots); + for _ in 0..slots { + let (k, v) = cache.read_slot(slot)?; + if is_empty_slot(&k, &v) { + return Ok(None); + } + if k == key { + return Ok(Some( + FILE_HEADER_LEN as u64 + + slot * SH_HEAD_SLOT_SIZE as u64 + + SH_HEAD_KEY_LEN as u64, + )); + } + slot = (slot + 1) & (slots - 1); + } + Ok(None) + } + + pub(crate) fn read_value_word(&self, off: u64) -> Result<[u8; 8], StoreError> { + let mut buf = [0u8; 8]; + self.file.read_at(off, &mut buf)?; + Ok(buf) + } + + pub(crate) fn write_value_word(&self, off: u64, bytes: &[u8; 8]) -> Result<(), StoreError> { + self.file.write_at(off, bytes) + } + /// Soft-clear value; keeps probe chain. pub fn clear_key(&self, full: &[u8; 32]) -> Result { let key = Self::to_key(full); diff --git a/crates/rbitcoin-store/src/scripthash_mphf.rs b/crates/rbitcoin-store/src/scripthash_mphf.rs index 3f012b5320..a925483bf6 100644 --- a/crates/rbitcoin-store/src/scripthash_mphf.rs +++ b/crates/rbitcoin-store/src/scripthash_mphf.rs @@ -203,6 +203,24 @@ impl MphfHead { Ok(true) } + /// Byte offset of the pack8 word in `.val`, when the key is present. + pub(crate) fn value_word_offset(&self, key: &ShHeadKey) -> Result, StoreError> { + Ok(self + .slot_if_present(key)? + .map(|slot| slot.saturating_mul(8))) + } + + pub(crate) fn read_value_word(&self, off: u64) -> Result<[u8; 8], StoreError> { + let mut buf = [0u8; 8]; + pread_file_exact(&self.val_file, off, &mut buf) + .map_err(|e| StoreError::io(val_path(&self.base), e))?; + Ok(buf) + } + + pub(crate) fn write_value_word(&self, off: u64, bytes: &[u8; 8]) -> Result<(), StoreError> { + pwrite_file(&self.val_file, off, bytes).map_err(|e| StoreError::io(val_path(&self.base), e)) + } + pub fn for_each_occupied( &self, mut f: impl FnMut(ShHeadKey, ShHeadValue) -> Result<(), StoreError>, diff --git a/crates/rbitcoin-store/src/scripthash_sorted_head.rs b/crates/rbitcoin-store/src/scripthash_sorted_head.rs index 95dc513755..f1c9f4f5ca 100644 --- a/crates/rbitcoin-store/src/scripthash_sorted_head.rs +++ b/crates/rbitcoin-store/src/scripthash_sorted_head.rs @@ -143,6 +143,26 @@ impl SortedHead { Ok(Some(unpack8_bytes(&val)?)) } + /// Byte offset of the pack8 word, when the key is on this file. + pub(crate) fn value_word_offset(&self, key: &ShHeadKey) -> Result, StoreError> { + let Some((slot, _)) = self.locate_rec(key)? else { + return Ok(None); + }; + Ok(Some( + DATA_HEADER_LEN + slot * SH_HEAD_SLOT_SIZE as u64 + SH_HEAD_KEY_LEN as u64, + )) + } + + pub(crate) fn read_value_word(&self, off: u64) -> Result<[u8; 8], StoreError> { + let mut buf = [0u8; 8]; + pread_file_exact(&self.file, off, &mut buf).map_err(|e| StoreError::io(&self.path, e))?; + Ok(buf) + } + + pub(crate) fn write_value_word(&self, off: u64, bytes: &[u8; 8]) -> Result<(), StoreError> { + pwrite_file(&self.file, off, bytes).map_err(|e| StoreError::io(&self.path, e)) + } + /// In-place `value16` update. `Ok(false)` if the key is not on this file. pub fn update_value(&self, key: &ShHeadKey, value: &ShHeadValue) -> Result { if let Some(ref fuse) = self.fuse { diff --git a/crates/rbitcoin-store/src/scripthash_tests.rs b/crates/rbitcoin-store/src/scripthash_tests.rs index 2ea7fcad05..f6356a5c7e 100644 --- a/crates/rbitcoin-store/src/scripthash_tests.rs +++ b/crates/rbitcoin-store/src/scripthash_tests.rs @@ -12,7 +12,8 @@ use crate::scripthash_materialize::{ use crate::scripthash_mphf::{mix_key16, MphfHead}; use crate::scripthash_pages::{ sh_page_as_array, sh_page_as_array_mut, sh_page_extent, sh_page_set_extent, - SH_PAGE_EXTENT_STREAM_MAX, SH_PAGE_SIZE, SH_PAGE_STREAM_MAX, + sh_page_set_extent_creates, sh_page_set_next, SH_PAGE_EXTENT_STREAM_MAX, SH_PAGE_OFF_N_FKS, + SH_PAGE_SIZE, SH_PAGE_STREAM_MAX, }; use std::sync::atomic::AtomicBool; @@ -1040,7 +1041,11 @@ fn freelist_reuses_page() { } other => panic!("expected slab, got {other:?}"), }; - assert_eq!(off1, off2, "slab freelist should reuse offset"); + let step = crate::scripthash_layout::slab_bytes(0); + assert!( + off2 == off1 || off2 == off1 + step, + "slab freelist should reuse a freed offset (first {off1}, next {off2})" + ); let _ = std::fs::remove_dir_all(&dir); } @@ -1892,6 +1897,564 @@ fn extent_append_glued_bumps_extent_n() { let _ = std::fs::remove_dir_all(&dir); } +/// A tip disconnect removes the highest creates. That must rewrite the tail +/// page, not allocate another copy of the whole extent. +#[test] +fn extent_suffix_unlink_does_not_copy_the_chain() { + let dir = tmp(); + let t = ScriptHashTable::create_tiny(&dir).unwrap(); + // Last page must keep the remaining creates. A one-fk tail would promote + // the previous full page and allocate a split, which is a different budget. + let n = (SH_PAGE_STREAM_MAX as u64).saturating_add(400); + let mut sh = [0u8; 32]; + sh[0] = 0x31; + let ents: Vec<_> = (1..=n).map(Fk).collect(); + let mut session = t.pack_shard_session(0).unwrap(); + push_chain(&mut session, sh, &ents); + seal_session(&t, 0, session); + let before = t.body().logical_len(); + assert!(t.unlink_create(&sh, Fk(n), 0).unwrap()); + let growth = t.body().logical_len().saturating_sub(before); + assert!( + growth <= SH_PAGE_SIZE as u64, + "suffix unlink grew the body by {growth} bytes (chain copy)" + ); + let got = t.entries(&sh).unwrap(); + assert_eq!(got.len(), (n - 1) as usize); + assert_eq!(got.last().map(|e| e.0), Some(Fk(n - 1))); + assert_eq!(got.first().map(|e| e.0), Some(Fk(1))); + let after = t.body().logical_len(); + assert!(!t.unlink_create(&sh, Fk(n), 0).unwrap()); + assert_eq!(t.body().logical_len(), after, "absent unlink must not grow"); + let _ = std::fs::remove_dir_all(&dir); +} + +/// Dropping more than one tail page still must not copy the prefix. +#[test] +fn extent_suffix_batch_unlink_stays_within_two_pages() { + let dir = tmp(); + let t = ScriptHashTable::create_tiny(&dir).unwrap(); + let n = (SH_PAGE_STREAM_MAX as u64).saturating_mul(2); + let mut sh = [0u8; 32]; + sh[0] = 0x32; + let ents: Vec<_> = (1..=n).map(Fk).collect(); + let mut session = t.pack_shard_session(0).unwrap(); + push_chain(&mut session, sh, &ents); + seal_session(&t, 0, session); + let drop_n = (SH_PAGE_STREAM_MAX as u64).saturating_add(50); + let fks: Vec<_> = ((n - drop_n + 1)..=n).map(Fk).collect(); + let before = t.body().logical_len(); + assert!(t.unlink_creates(&sh, &fks).unwrap()); + let growth = t.body().logical_len().saturating_sub(before); + assert!( + growth <= (SH_PAGE_SIZE as u64).saturating_mul(2), + "batch suffix unlink grew the body by {growth} bytes" + ); + let got = t.entries(&sh).unwrap(); + assert_eq!(got.len(), (n - drop_n) as usize); + assert_eq!(got.last().map(|e| e.0), Some(Fk(n - drop_n))); + assert!(!t.unlink_creates(&sh, &fks).unwrap()); + let again = t.entries(&sh).unwrap(); + assert_eq!(again.len(), got.len()); + let _ = std::fs::remove_dir_all(&dir); +} + +/// A torn tail on the ingest home restores the page and the ingest head word. +#[test] +fn extent_ingest_torn_tail_restores_on_open() { + let dir = tmp(); + let sh = script_hash(&[0x36]); + let n = 300u64; + { + let t = ScriptHashTable::create_tiny(&dir).unwrap(); + let many: Vec<_> = (1..=n).map(|i| rec(sh, i, 0)).collect(); + t.put_create_batch_append(&many, &mut HashMap::new()) + .unwrap(); + assert!(matches!(t.key_home(&sh).unwrap(), KeyHome::Ingest)); + t.test_tear_extent_tail(&sh).unwrap(); + } + let t = ScriptHashTable::open_tiny(&dir).unwrap(); + let got = t.entries(&sh).unwrap(); + assert_eq!(got.len(), n as usize); + assert_eq!(got.first().map(|e| e.0), Some(Fk(1))); + assert_eq!(got.last().map(|e| e.0), Some(Fk(n))); + let _ = std::fs::remove_dir_all(&dir); +} + +/// A torn tail on a sealed overflow file, and again after L1 promote. +#[test] +fn extent_sealed_ovf_torn_tail_restores_on_open() { + let dir = tmp(); + let n = 300u64; + let mut sh = [0u8; 32]; + sh[0] = 0x37; + { + let t = ScriptHashTable::create_tiny(&dir).unwrap(); + let sh_main = script_hash(&[0x10]); + let mut session = t.pack_shard_session(0).unwrap(); + push_chain(&mut session, sh_main, &[Fk(1)]); + seal_session(&t, 0, session); + let many: Vec<_> = (1..=n).map(|i| rec(sh, i, 0)).collect(); + t.put_create_batch_append(&many, &mut HashMap::new()) + .unwrap(); + for i in 0..210u32 { + let filler = script_hash(&[0xa4, (i & 0xff) as u8, (i >> 8) as u8, 0x04]); + put_create(&t, rec(filler, 5000 + u64::from(i), 0)); + } + assert!(matches!(t.key_home(&sh).unwrap(), KeyHome::SealedOvf)); + t.test_tear_extent_tail(&sh).unwrap(); + t.flush().unwrap(); + } + { + let t = ScriptHashTable::open_tiny(&dir).unwrap(); + assert_eq!(t.entries(&sh).unwrap().len(), n as usize); + for i in 0..210u32 { + let filler = script_hash(&[0xa5, (i & 0xff) as u8, (i >> 8) as u8, 0x05]); + put_create(&t, rec(filler, 8000 + u64::from(i), 0)); + } + t.compact_sealed_ovf().unwrap(); + assert!(t.ovf_l1.lock().unwrap().is_some()); + assert!(matches!(t.key_home(&sh).unwrap(), KeyHome::SealedOvf)); + t.test_tear_extent_tail(&sh).unwrap(); + t.flush().unwrap(); + } + let t = ScriptHashTable::open_tiny(&dir).unwrap(); + let got = t.entries(&sh).unwrap(); + assert_eq!(got.len(), n as usize); + assert_eq!(got.last().map(|e| e.0), Some(Fk(n))); + let _ = std::fs::remove_dir_all(&dir); +} + +/// Dropping a linked tail page finds the previous page by walking `next`. +#[test] +fn extent_linked_tail_unlink_walks_to_the_extent() { + let dir = tmp(); + let t = ScriptHashTable::create_tiny(&dir).unwrap(); + let n = SH_PAGE_STREAM_MAX + SH_PAGE_EXTENT_STREAM_MAX - 1; + let mut sh = [0u8; 32]; + sh[0] = 0x38; + let ents: Vec<_> = (1..=n as u64).map(Fk).collect(); + let mut sh_gap = [0u8; 32]; + sh_gap[0] = 0x38; + sh_gap[1] = 0x01; + let mut session = t.pack_shard_session(0).unwrap(); + push_chain(&mut session, sh, &ents); + push_chain(&mut session, sh_gap, &[Fk(1), Fk(2)]); + seal_session(&t, 0, session); + let (base, last0, n0) = extent_meta(&t, &sh); + assert_eq!(n0, 2); + put_create(&t, rec(sh, n as u64 + 1, 0)); + let (_, last1, _) = extent_meta(&t, &sh); + assert_ne!(last1, base + u64::from(n0 - 1) * SH_PAGE_SIZE as u64); + assert_ne!(last1, last0); + assert!(t.unlink_create(&sh, Fk(n as u64 + 1), 0).unwrap()); + let got = t.entries(&sh).unwrap(); + assert_eq!(got.len(), n); + assert_eq!(got.last().map(|e| e.0), Some(Fk(n as u64))); + let _ = std::fs::remove_dir_all(&dir); +} + +/// A torn tail write rolls back to the pre-unlink page on open. +#[test] +fn extent_torn_tail_restores_on_open() { + let dir = tmp(); + let n = (SH_PAGE_STREAM_MAX as u64).saturating_add(10); + let mut sh = [0u8; 32]; + sh[0] = 0x33; + { + let t = ScriptHashTable::create_tiny(&dir).unwrap(); + let ents: Vec<_> = (1..=n).map(Fk).collect(); + let mut session = t.pack_shard_session(0).unwrap(); + push_chain(&mut session, sh, &ents); + seal_session(&t, 0, session); + t.test_tear_extent_tail(&sh).unwrap(); + } + let t = ScriptHashTable::open_tiny(&dir).unwrap(); + let got = t.entries(&sh).unwrap(); + assert_eq!(got.len(), n as usize); + assert_eq!(got.first().map(|e| e.0), Some(Fk(1))); + assert_eq!(got.last().map(|e| e.0), Some(Fk(n))); + assert!( + matches!( + t.head_value(&sh).unwrap().unwrap(), + ShHeadValue::Extent { .. } + ), + "open restores the head word, not the empty word written after the tear" + ); + let _ = std::fs::remove_dir_all(&dir); +} + +/// Removing a create that is not the tail still leaves a readable key. +#[test] +fn extent_middle_unlink_roundtrips() { + let dir = tmp(); + let n = (SH_PAGE_STREAM_MAX as u64).saturating_mul(2); + let mut sh = [0u8; 32]; + sh[0] = 0x34; + { + let t = ScriptHashTable::create_tiny(&dir).unwrap(); + let ents: Vec<_> = (1..=n).map(Fk).collect(); + let mut session = t.pack_shard_session(0).unwrap(); + push_chain(&mut session, sh, &ents); + seal_session(&t, 0, session); + assert!(t.unlink_create(&sh, Fk(10), 0).unwrap()); + let got = t.entries(&sh).unwrap(); + assert_eq!(got.len(), (n - 1) as usize); + assert!(got.iter().all(|e| e.0 != Fk(10))); + t.flush().unwrap(); + } + let t = ScriptHashTable::open_tiny(&dir).unwrap(); + let got = t.entries(&sh).unwrap(); + assert_eq!(got.len(), (n - 1) as usize); + assert!(got.iter().all(|e| e.0 != Fk(10))); + let _ = std::fs::remove_dir_all(&dir); +} + +#[test] +fn append_rejects_extent_page_that_is_not_ver2() { + let dir = tmp(); + let t = ScriptHashTable::create_tiny(&dir).unwrap(); + let n = SH_PAGE_STREAM_MAX as u64; + let mut sh = [0u8; 32]; + sh[0] = 0x35; + let ents: Vec<_> = (1..=n).map(Fk).collect(); + let mut session = t.pack_shard_session(0).unwrap(); + push_chain(&mut session, sh, &ents); + seal_session(&t, 0, session); + let last_page = match t.head_value(&sh).unwrap().unwrap() { + ShHeadValue::Extent { last_page } => last_page, + other => panic!("expected extent, got {other:?}"), + }; + let mut page = [0u8; SH_PAGE_SIZE]; + t.body().read_at(last_page, &mut page).unwrap(); + page[0] = 0; + t.body().write_at(last_page, &page).unwrap(); + let err = t + .put_create_batch_append(&[rec(sh, n + 1, 0)], &mut HashMap::new()) + .unwrap_err(); + let msg = err.to_string(); + assert!( + msg.contains("scripthash extent last page not ver=2"), + "{msg}" + ); + let _ = std::fs::remove_dir_all(&dir); +} + +/// Dropping a whole tail page must not break a history read that still +/// names the old head. +#[test] +fn extent_whole_page_drop_stays_readable() { + let dir = tmp(); + let t = ScriptHashTable::create_tiny(&dir).unwrap(); + let n = (SH_PAGE_STREAM_MAX as u64).saturating_mul(2); + let mut sh = [0u8; 32]; + sh[0] = 0x39; + let ents: Vec<_> = (1..=n).map(Fk).collect(); + let mut session = t.pack_shard_session(0).unwrap(); + push_chain(&mut session, sh, &ents); + seal_session(&t, 0, session); + t.watch_readable_during_unlink(true); + assert!(t.unlink_create(&sh, Fk(n), 0).unwrap()); + assert!( + !t.readable_broke(), + "a reader of the published head saw a broken extent" + ); + t.watch_readable_during_unlink(false); + let got = t.entries(&sh).unwrap(); + assert_eq!(got.len(), (n - 1) as usize); + assert_eq!(got.last().map(|e| e.0), Some(Fk(n - 1))); + let _ = std::fs::remove_dir_all(&dir); +} + +/// A failed undo clear must put the tail back. The next unlink then drops it. +#[test] +fn extent_undo_clear_failure_keeps_the_old_tail() { + let dir = tmp(); + let n = (SH_PAGE_STREAM_MAX as u64).saturating_add(400); + let mut sh = [0u8; 32]; + sh[0] = 0x3a; + let t = ScriptHashTable::create_tiny(&dir).unwrap(); + let ents: Vec<_> = (1..=n).map(Fk).collect(); + let mut session = t.pack_shard_session(0).unwrap(); + push_chain(&mut session, sh, &ents); + seal_session(&t, 0, session); + t.fail_next_undo_clear(); + let err = t.unlink_create(&sh, Fk(n), 0).unwrap_err(); + assert!( + err.to_string().contains("scripthash undo clear failed"), + "{err}" + ); + assert_eq!(t.entries(&sh).unwrap().len(), n as usize); + assert!(t.unlink_create(&sh, Fk(n), 0).unwrap()); + assert_eq!(t.entries(&sh).unwrap().len(), (n - 1) as usize); + t.flush().unwrap(); + drop(t); + let t = ScriptHashTable::open_tiny(&dir).unwrap(); + assert_eq!(t.entries(&sh).unwrap().len(), (n - 1) as usize); + let _ = std::fs::remove_dir_all(&dir); +} + +/// `extent_n` past the real chain is a hint. The ver=2 page is still the tail. +#[test] +fn append_accepts_inflated_extent_tail() { + let dir = tmp(); + let t = ScriptHashTable::create_tiny(&dir).unwrap(); + let sh = script_hash(&[0x7d]); + let mut heads = HashMap::new(); + let n = 600u32; + let many: Vec<_> = (1..=n).map(|v| rec(sh, u64::from(v), v)).collect(); + t.put_create_batch_append(&many, &mut heads).unwrap(); + let last_page = match t.head_value(&sh).unwrap().unwrap() { + ShHeadValue::Extent { last_page } => last_page, + other => panic!("expected extent, got {other:?}"), + }; + let mut page = [0u8; SH_PAGE_SIZE]; + t.ovf_file().read_at(last_page, &mut page).unwrap(); + let arr = sh_page_as_array_mut(&mut page).unwrap(); + let (base, pages) = sh_page_extent(arr).unwrap().expect("ver=2 last page"); + let over = (64 * 1024 * 1024 / SH_PAGE_SIZE as u64) as u32 + 1; + assert!(over > pages); + sh_page_set_extent(arr, base, over).unwrap(); + t.ovf_file().write_at(last_page, &page).unwrap(); + put_create(&t, rec(sh, u64::from(n) + 1, 0)); + let got = t.entries(&sh).unwrap(); + assert_eq!(got.len(), n as usize + 1); + assert_eq!(got.last().map(|e| e.0), Some(Fk(u64::from(n) + 1))); + let _ = std::fs::remove_dir_all(&dir); +} + +/// A linked tail popped from the page freelist may sit below the extent. +#[test] +fn append_accepts_linked_tail_below_the_extent() { + let dir = tmp(); + let t = ScriptHashTable::create_tiny(&dir).unwrap(); + let n = SH_PAGE_STREAM_MAX + SH_PAGE_EXTENT_STREAM_MAX - 1; + let mut sh_a = [0u8; 32]; + sh_a[0] = 0x51; + let mut sh = [0u8; 32]; + sh[0] = 0x52; + let mut sh_gap = [0u8; 32]; + sh_gap[0] = 0x53; + let a_ents: Vec<_> = (1..=SH_PAGE_STREAM_MAX as u64).map(Fk).collect(); + let ents: Vec<_> = (1..=n as u64).map(Fk).collect(); + let mut session = t.pack_shard_session(0).unwrap(); + push_chain(&mut session, sh_a, &a_ents); + push_chain(&mut session, sh, &ents); + push_chain(&mut session, sh_gap, &[Fk(1), Fk(2)]); + seal_session(&t, 0, session); + let a_page = match t.head_value(&sh_a).unwrap().unwrap() { + ShHeadValue::Extent { last_page } => last_page, + other => panic!("expected extent for the low key, got {other:?}"), + }; + let (base, _, pages) = extent_meta(&t, &sh); + let end = base + u64::from(pages - 1) * SH_PAGE_SIZE as u64; + assert!( + a_page < end, + "freed page {a_page} must sit below the extent end {end}" + ); + assert!(t.unlink_creates(&sh_a, &a_ents).unwrap()); + put_create(&t, rec(sh, n as u64 + 1, 0)); + let (_, last, _) = extent_meta(&t, &sh); + assert!( + last < end, + "linked tail {last} should sit below the stamped end {end} (freed page {a_page})" + ); + put_create(&t, rec(sh, n as u64 + 2, 0)); + assert_eq!(t.entries(&sh).unwrap().len(), n + 2); + let _ = std::fs::remove_dir_all(&dir); +} + +fn zero_n_fks(body: &crate::file::TableFile, off: u64) { + let mut page = [0u8; SH_PAGE_SIZE]; + body.read_at(off, &mut page).unwrap(); + page[SH_PAGE_OFF_N_FKS..SH_PAGE_OFF_N_FKS + 2].copy_from_slice(&0u16.to_le_bytes()); + body.write_at(off, &page).unwrap(); +} + +/// A head page emptied before the predecessor is rewritten is still the tail. +/// Unlink walks it and drops the suffix from the page that still holds creates. +#[test] +fn extent_empty_head_unlink_walks_to_the_predecessor() { + let dir = tmp(); + let t = ScriptHashTable::create_tiny(&dir).unwrap(); + let n = (SH_PAGE_STREAM_MAX as u64).saturating_mul(2); + let mut sh = [0u8; 32]; + sh[0] = 0x3b; + let ents: Vec<_> = (1..=n).map(Fk).collect(); + let mut session = t.pack_shard_session(0).unwrap(); + push_chain(&mut session, sh, &ents); + seal_session(&t, 0, session); + let (base, last, _) = extent_meta(&t, &sh); + assert_ne!(last, base, "the head page must have a predecessor"); + zero_n_fks(t.body(), last); + let before = t.entries(&sh).unwrap(); + assert!(!before.is_empty(), "the predecessor still holds creates"); + let drop_fk = before.last().unwrap().0; + assert!(t.unlink_create(&sh, drop_fk, 0).unwrap()); + let got = t.entries(&sh).unwrap(); + assert_eq!(got.len(), before.len() - 1); + assert!(got.iter().all(|e| e.0 != drop_fk)); + assert_eq!(got.last().map(|e| e.0), Some(Fk(drop_fk.0 - 1))); + let _ = std::fs::remove_dir_all(&dir); +} + +/// The tail's creates are gone and the only earlier page is empty, so the key goes. +#[test] +fn extent_empty_predecessor_drops_the_key() { + let dir = tmp(); + let t = ScriptHashTable::create_tiny(&dir).unwrap(); + let n = (SH_PAGE_STREAM_MAX as u64).saturating_mul(2); + let mut sh = [0u8; 32]; + sh[0] = 0x3c; + let ents: Vec<_> = (1..=n).map(Fk).collect(); + let mut session = t.pack_shard_session(0).unwrap(); + push_chain(&mut session, sh, &ents); + seal_session(&t, 0, session); + let (base, last, _) = extent_meta(&t, &sh); + assert_ne!(last, base); + zero_n_fks(t.body(), base); + let tail = t.entries(&sh).unwrap(); + assert!(!tail.is_empty()); + assert!(tail.iter().all(|e| e.0 .0 > 1)); + let mut remove: Vec = tail.iter().map(|e| e.0).collect(); + remove.push(Fk(1)); + assert!(t.unlink_creates(&sh, &remove).unwrap()); + assert!(t.entries(&sh).unwrap().is_empty()); + let _ = std::fs::remove_dir_all(&dir); +} + +/// An empty one-page extent has no create to drop. +#[test] +fn extent_empty_only_page_unlink_is_absent() { + let dir = tmp(); + let t = ScriptHashTable::create_tiny(&dir).unwrap(); + let n = 2100u64; + let mut sh = [0u8; 32]; + sh[0] = 0x3d; + let ents: Vec<_> = (1..=n).map(Fk).collect(); + let mut session = t.pack_shard_session(0).unwrap(); + push_chain(&mut session, sh, &ents); + seal_session(&t, 0, session); + let (base, last, pages) = extent_meta(&t, &sh); + assert_eq!(pages, 1); + assert_eq!(last, base); + zero_n_fks(t.body(), last); + assert!(!t.unlink_create(&sh, Fk(1), 0).unwrap()); + assert!(t.entries(&sh).unwrap().is_empty()); + let _ = std::fs::remove_dir_all(&dir); +} + +/// A page that still names a successor is not the tail. +#[test] +fn append_rejects_a_page_that_still_has_a_successor() { + let dir = tmp(); + let t = ScriptHashTable::create_tiny(&dir).unwrap(); + let n = 2100u64; + let mut sh = [0u8; 32]; + sh[0] = 0x3e; + let ents: Vec<_> = (1..=n).map(Fk).collect(); + let mut session = t.pack_shard_session(0).unwrap(); + push_chain(&mut session, sh, &ents); + seal_session(&t, 0, session); + let last = match t.head_value(&sh).unwrap().unwrap() { + ShHeadValue::Extent { last_page } => last_page, + other => panic!("expected extent, got {other:?}"), + }; + let mut page = [0u8; SH_PAGE_SIZE]; + t.body().read_at(last, &mut page).unwrap(); + sh_page_set_next(sh_page_as_array_mut(&mut page).unwrap(), last).unwrap(); + t.body().write_at(last, &page).unwrap(); + let err = t + .put_create_batch_append(&[rec(sh, n + 1, 0)], &mut HashMap::new()) + .unwrap_err(); + assert!( + err.to_string() + .contains("scripthash extent last page is not the extent tail"), + "{err}" + ); + let _ = std::fs::remove_dir_all(&dir); +} + +/// A zero creates counter is recounted from the pages before the new create lands. +#[test] +fn append_recounts_when_the_creates_counter_is_zero() { + let dir = tmp(); + let t = ScriptHashTable::create_tiny(&dir).unwrap(); + let n = 2100u64; + let mut sh = [0u8; 32]; + sh[0] = 0x3f; + let ents: Vec<_> = (1..=n).map(Fk).collect(); + let mut session = t.pack_shard_session(0).unwrap(); + push_chain(&mut session, sh, &ents); + seal_session(&t, 0, session); + let last = match t.head_value(&sh).unwrap().unwrap() { + ShHeadValue::Extent { last_page } => last_page, + other => panic!("expected extent, got {other:?}"), + }; + let mut page = [0u8; SH_PAGE_SIZE]; + t.body().read_at(last, &mut page).unwrap(); + sh_page_set_extent_creates(sh_page_as_array_mut(&mut page).unwrap(), 0); + t.body().write_at(last, &page).unwrap(); + put_create(&t, rec(sh, n + 1, 0)); + let got = t.entries(&sh).unwrap(); + assert_eq!(got.len(), n as usize + 1); + assert_eq!(got.last().map(|e| e.0), Some(Fk(n + 1))); + let _ = std::fs::remove_dir_all(&dir); +} + +/// An empty tail whose creates counter is also zero still accepts one create. +#[test] +fn append_accepts_an_empty_tail_with_a_zero_creates_counter() { + let dir = tmp(); + let t = ScriptHashTable::create_tiny(&dir).unwrap(); + let n = 2100u64; + let mut sh = [0u8; 32]; + sh[0] = 0x40; + let ents: Vec<_> = (1..=n).map(Fk).collect(); + let mut session = t.pack_shard_session(0).unwrap(); + push_chain(&mut session, sh, &ents); + seal_session(&t, 0, session); + let last = match t.head_value(&sh).unwrap().unwrap() { + ShHeadValue::Extent { last_page } => last_page, + other => panic!("expected extent, got {other:?}"), + }; + zero_n_fks(t.body(), last); + let mut page = [0u8; SH_PAGE_SIZE]; + t.body().read_at(last, &mut page).unwrap(); + sh_page_set_extent_creates(sh_page_as_array_mut(&mut page).unwrap(), 0); + t.body().write_at(last, &page).unwrap(); + put_create(&t, rec(sh, n + 1, 0)); + let got = t.entries(&sh).unwrap(); + assert_eq!(got.len(), 1); + assert_eq!(got[0].0, Fk(n + 1)); + let _ = std::fs::remove_dir_all(&dir); +} + +/// Two slab unlinks publish both heads, then free. Reopen sees both. +#[test] +fn unlink_batch_publishes_both_heads() { + let dir = tmp(); + let t = ScriptHashTable::create_tiny(&dir).unwrap(); + let a = script_hash(&[0x61]); + let b = script_hash(&[0x62]); + for i in 1..=3u64 { + put_create(&t, rec(a, i, i as u32)); + put_create(&t, rec(b, i + 10, i as u32)); + } + t.unlink_creates_batch(&[(a, vec![Fk(2)]), (b, vec![Fk(12)])]) + .unwrap(); + assert_eq!(t.entries(&a).unwrap().len(), 2); + assert!(t.entries(&a).unwrap().iter().all(|e| e.0 != Fk(2))); + assert_eq!(t.entries(&b).unwrap().len(), 2); + assert!(t.entries(&b).unwrap().iter().all(|e| e.0 != Fk(12))); + t.flush().unwrap(); + drop(t); + let t = ScriptHashTable::open_tiny(&dir).unwrap(); + assert_eq!(t.entries(&a).unwrap().len(), 2); + assert_eq!(t.entries(&b).unwrap().len(), 2); + let _ = std::fs::remove_dir_all(&dir); +} + #[test] fn reinit_clears_head_when_live_count_already_zero() { // Crash mid-finish: heads durable, alloc live_count still 0. diff --git a/crates/rbitcoin-store/src/scripthash_unlink.rs b/crates/rbitcoin-store/src/scripthash_unlink.rs new file mode 100644 index 0000000000..a0f0ad26f6 --- /dev/null +++ b/crates/rbitcoin-store/src/scripthash_unlink.rs @@ -0,0 +1,1072 @@ +//! Disconnect unlink for one scripthash. +//! +//! Tip creates are a suffix of the key. An extent drops that suffix from the +//! tail pages. The old page image is saved first, then the new tail is written, +//! then the head. A crash before the undo record is cleared restores the old +//! tail and the old head word on open. + +use super::{ + sh_page_as_array, sh_page_chunk_ranges, sh_page_extent, sh_page_next, + sh_page_pack_extent_last_fks, sh_page_pack_fks, write_alloc_header, AllocState, LocatedHome, + ScriptHashTable, StoreError, TableFile, SH_PAGE_SIZE, SH_PAGE_SLAB_CLASS, +}; +use crate::scripthash_layout::ShHeadValue; +use crate::scripthash_pages::{ + sh_page_entries_into, sh_page_extent_creates, sh_page_set_extent_creates, +}; +use rbitcoin_primitives::Fk; +#[cfg(test)] +use std::cell::Cell; +use std::collections::BTreeSet; + +#[cfg(test)] +thread_local! { + static WATCH_READABLE: Cell = const { Cell::new(false) }; + static READABLE_BROKE: Cell = const { Cell::new(false) }; + static FAIL_CLEAR_UNDO: Cell = const { Cell::new(false) }; +} + +const UNDO_AT: usize = 256; +const UNDO_LEN: usize = 48; +const UNDO_MAGIC: [u8; 4] = *b"SHUN"; + +const HEAD_NONE: u8 = 0; +const HEAD_MAIN: u8 = 1; +const HEAD_INGEST: u8 = 2; +const HEAD_SORTED_OVF: u8 = 3; +const HEAD_OVF_L1: u8 = 4; + +struct HeadWord { + tag: u8, + index: u16, + off: u64, + old: [u8; 8], +} + +struct UndoRec { + target: u64, + image: u64, + word: Option, +} + +#[derive(Default)] +struct UnlinkBatch { + rewrites: Vec, + main_shards: Vec, + ovf: bool, +} + +struct DeferredRewrite { + scripthash: [u8; 32], + home: LocatedHome, + new_val: ShHeadValue, + old: ShHeadValue, + removed: u64, +} + +enum SuffixPlan { + NotSuffix, + Absent, + DropAll { + removed: u64, + }, + Rewrite { + boundary: u64, + keep: Vec, + drop_pages: Vec, + base: u64, + old_n: u32, + creates: u32, + removed: u64, + }, +} + +impl ScriptHashTable { + /// Remove `create_tx_fk`s from one scripthash. Duplicate fks count once. + /// + /// `Ok(true)` when at least one fk was stored. A pure suffix of an extent + /// rewrites the tail pages. Any other shape rewrites the key once, and the + /// previous pages stay readable until the head names the replacement. + pub fn unlink_creates( + &self, + scripthash: &[u8; 32], + create_tx_fks: &[Fk], + ) -> Result { + let mut batch = UnlinkBatch::default(); + let changed = self.unlink_one(scripthash, create_tx_fks, &mut batch)?; + self.finish_unlink_batch(batch)?; + Ok(changed) + } + + /// Unlink every group, then sync the new chains and heads once before any + /// replaced page is freed. One scripthash uses [`Self::unlink_creates`]. + pub fn unlink_creates_batch(&self, groups: &[([u8; 32], Vec)]) -> Result<(), StoreError> { + let mut batch = UnlinkBatch::default(); + for (scripthash, fks) in groups { + self.unlink_one(scripthash, fks, &mut batch)?; + } + self.finish_unlink_batch(batch) + } + + fn unlink_one( + &self, + scripthash: &[u8; 32], + create_tx_fks: &[Fk], + batch: &mut UnlinkBatch, + ) -> Result { + let mut remove = BTreeSet::new(); + for fk in create_tx_fks { + if !fk.is_null() { + remove.insert(fk.0); + } + } + if remove.is_empty() { + return Ok(false); + } + let Some((val, home)) = self.locate_head(scripthash)? else { + return Ok(false); + }; + if val.is_empty() { + return Ok(false); + } + let ShHeadValue::Extent { last_page } = val else { + return self.unlink_by_rewrite(scripthash, home, &val, &remove, batch); + }; + let body = self.body_for(scripthash, super::KeyHome::from(home)); + match plan_extent_suffix(body, last_page, &remove)? { + SuffixPlan::Absent => Ok(false), + SuffixPlan::NotSuffix => self.unlink_by_rewrite(scripthash, home, &val, &remove, batch), + plan => self.apply_suffix_plan(scripthash, home, &val, plan), + } + } + + /// Unlink one create_tx_fk (disconnect tip). One tx pays a scripthash once, + /// however many outputs share that script. + pub fn unlink_create( + &self, + scripthash: &[u8; 32], + create_tx_fk: Fk, + _vout: u32, + ) -> Result { + self.unlink_creates(scripthash, &[create_tx_fk]) + } + + pub(super) fn recover_unlink_undos(&self) -> Result<(), StoreError> { + for (i, body) in self.bodies.iter().enumerate() { + self.recover_one_undo(body, &self.allocs[i])?; + } + if let (Some(body), Some(mu)) = (&self.ovf_body, &self.ovf_alloc) { + self.recover_one_undo(body, mu)?; + } + Ok(()) + } + + /// Leave a torn last page and an empty head word. Open restores both. + #[cfg(test)] + pub(super) fn test_tear_extent_tail(&self, scripthash: &[u8; 32]) -> Result<(), StoreError> { + let Some((val, home)) = self.locate_head(scripthash)? else { + return Err(StoreError::Corrupt("scripthash tear: missing key")); + }; + let ShHeadValue::Extent { last_page } = val else { + return Err(StoreError::Corrupt("scripthash tear: not an extent")); + }; + let body = self.body_for(scripthash, super::KeyHome::from(home)); + let alloc_mu = self.alloc_for(scripthash, super::KeyHome::from(home)); + let mut alloc = alloc_mu.lock().unwrap(); + let word = self.capture_head_word(scripthash, home)?; + let rec = arm_undo(self, body, &mut alloc, last_page, word)?; + let mut torn = [0u8; SH_PAGE_SIZE]; + body.read_at(last_page, &mut torn)?; + torn[0] = 0; + body.write_at(last_page, &torn)?; + let _ = rec; + self.unlink_write_home(scripthash, home, &ShHeadValue::Empty)?; + self.sync_home(home, scripthash)?; + body.sync_data_only()?; + Ok(()) + } + + #[cfg(test)] + pub(super) fn watch_readable_during_unlink(&self, on: bool) { + WATCH_READABLE.with(|c| c.set(on)); + if on { + READABLE_BROKE.with(|c| c.set(false)); + } + } + + #[cfg(test)] + pub(super) fn readable_broke(&self) -> bool { + READABLE_BROKE.with(|c| c.get()) + } + + #[cfg(test)] + pub(super) fn fail_next_undo_clear(&self) { + FAIL_CLEAR_UNDO.with(|c| c.set(true)); + } + + fn unlink_by_rewrite( + &self, + scripthash: &[u8; 32], + home: LocatedHome, + old: &ShHeadValue, + remove: &BTreeSet, + batch: &mut UnlinkBatch, + ) -> Result { + let body = self.body_for(scripthash, super::KeyHome::from(home)); + let mut live = self.collect_entries_from(body, old)?; + let before = live.len(); + live.retain(|fk| !remove.contains(&fk.0)); + if live.len() == before { + return Ok(false); + } + let alloc_mu = self.alloc_for(scripthash, super::KeyHome::from(home)); + let mut alloc = alloc_mu.lock().unwrap(); + let saved = *alloc; + let new_val = match self.rewrite_entries_for_key(body, &mut alloc, &live) { + Ok(v) => v, + Err(e) => { + *alloc = saved; + return Err(e); + } + }; + if let Err(e) = write_alloc_header(body, &alloc) { + *alloc = saved; + return Err(e); + } + let removed = (before - live.len()) as u64; + if frees_old_pages(old) { + // Head stays on the old chain until the batch syncs these bytes. + note_batch_body(self, batch, scripthash, home); + batch.rewrites.push(DeferredRewrite { + scripthash: *scripthash, + home, + new_val, + old: old.clone(), + removed, + }); + return Ok(true); + } + if let Err(e) = self.unlink_write_home(scripthash, home, &new_val) { + *alloc = saved; + let _ = write_alloc_header(body, &alloc); + return Err(e); + } + alloc.live_count = alloc.live_count.saturating_sub(removed); + write_alloc_header(body, &alloc)?; + Ok(true) + } + + fn finish_unlink_batch(&self, batch: UnlinkBatch) -> Result<(), StoreError> { + if batch.rewrites.is_empty() { + return Ok(()); + } + for si in &batch.main_shards { + self.bodies[*si].sync_data_only()?; + } + if batch.ovf { + if let Some(body) = &self.ovf_body { + body.sync_data_only()?; + } + } + for job in &batch.rewrites { + self.unlink_write_home(&job.scripthash, job.home, &job.new_val)?; + } + self.sync_batch_homes(&batch)?; + for job in &batch.rewrites { + let body = self.body_for(&job.scripthash, super::KeyHome::from(job.home)); + let alloc_mu = self.alloc_for(&job.scripthash, super::KeyHome::from(job.home)); + let mut alloc = alloc_mu.lock().unwrap(); + alloc.live_count = alloc.live_count.saturating_sub(job.removed); + if let Err(e) = self.free_if_paged(body, &mut alloc, &job.old) { + let _ = write_alloc_header(body, &alloc); + return Err(e); + } + write_alloc_header(body, &alloc)?; + } + Ok(()) + } + + fn sync_batch_homes(&self, batch: &UnlinkBatch) -> Result<(), StoreError> { + for si in &batch.main_shards { + if let Some(slot) = self.sorted_main.get(*si) { + if let Some(h) = slot.read().unwrap().as_ref() { + h.flush()?; + } + } + } + let ingest = batch + .rewrites + .iter() + .any(|job| matches!(job.home, LocatedHome::Main | LocatedHome::Ingest)); + let ovf = batch + .rewrites + .iter() + .any(|job| matches!(job.home, LocatedHome::SealedOvf)); + if ingest { + self.ingest.lock().unwrap().flush()?; + } + if ovf { + for h in self.sealed_ovf.lock().unwrap().iter() { + h.flush()?; + } + if let Some(l1) = self.ovf_l1.lock().unwrap().as_ref() { + l1.head.flush()?; + } + } + Ok(()) + } + + fn apply_suffix_plan( + &self, + scripthash: &[u8; 32], + home: LocatedHome, + old: &ShHeadValue, + plan: SuffixPlan, + ) -> Result { + let body = self.body_for(scripthash, super::KeyHome::from(home)); + let alloc_mu = self.alloc_for(scripthash, super::KeyHome::from(home)); + let mut alloc = alloc_mu.lock().unwrap(); + match plan { + SuffixPlan::DropAll { removed } => { + self.unlink_write_home(scripthash, home, &ShHeadValue::Empty)?; + self.sync_home(home, scripthash)?; + self.free_if_paged(body, &mut alloc, old)?; + alloc.live_count = alloc.live_count.saturating_sub(removed); + write_alloc_header(body, &alloc)?; + Ok(removed > 0) + } + SuffixPlan::Rewrite { + boundary, + keep, + drop_pages, + base, + old_n, + creates, + removed, + } => { + let old_last = match old { + ShHeadValue::Extent { last_page } => *last_page, + _ => boundary, + }; + let extent_n = new_extent_n(base, old_n, boundary); + let creates = new_creates(creates, removed); + if boundary != old_last { + // The head still names `old_last`. Shrink that page's + // extent first so `boundary` is not an interior page + // when it is rewritten. + self.retire_head_page( + body, + &mut alloc, + scripthash, + home, + old_last, + &BoundaryPack { + base, + extent_n, + boundary, + creates, + }, + )?; + } + let prepared = prepare_boundary( + self, + body, + &mut alloc, + &BoundaryPack { + base, + extent_n, + boundary, + creates, + }, + &keep, + )?; + let new_head = if prepared.new_last == old_last && drop_pages.is_empty() { + None + } else { + Some(ShHeadValue::extent(prepared.new_last)) + }; + self.commit_tail_page( + body, + &mut alloc, + scripthash, + home, + TailWrite { + target: prepared.at, + bytes: &prepared.bytes, + new_head, + head_page: old_last, + removed, + }, + )?; + for off in drop_pages { + self.free_slab(body, &mut alloc, SH_PAGE_SLAB_CLASS, off)?; + } + write_alloc_header(body, &alloc)?; + Ok(true) + } + SuffixPlan::Absent | SuffixPlan::NotSuffix => Ok(false), + } + } + + fn commit_tail_page( + &self, + body: &TableFile, + alloc: &mut AllocState, + scripthash: &[u8; 32], + home: LocatedHome, + write: TailWrite<'_>, + ) -> Result<(), StoreError> { + let word = if write.new_head.is_some() { + self.capture_head_word(scripthash, home)? + } else { + None + }; + let rec = arm_undo(self, body, alloc, write.target, word)?; + if let Err(e) = body.write_at(write.target, write.bytes) { + self.rollback_undo(body, alloc, &rec)?; + return Err(e); + } + body.sync_data_only()?; + note_readable(body, write.head_page); + if let Some(val) = write.new_head { + if let Err(e) = self.unlink_write_home(scripthash, home, &val) { + self.rollback_undo(body, alloc, &rec)?; + return Err(e); + } + if let Err(e) = self.sync_home(home, scripthash) { + self.rollback_undo(body, alloc, &rec)?; + return Err(e); + } + if let ShHeadValue::Extent { last_page } = val { + note_readable(body, last_page); + } + } + alloc.live_count = alloc.live_count.saturating_sub(write.removed); + if let Err(e) = clear_undo_keep_image(body, alloc) { + alloc.live_count = alloc.live_count.saturating_add(write.removed); + self.rollback_undo(body, alloc, &rec)?; + return Err(e); + } + self.free_slab(body, alloc, SH_PAGE_SLAB_CLASS, rec.image)?; + Ok(()) + } + + /// Publish a shorter extent on the page the head already names. + /// + /// The page stays `ver=2` and empty. Readers follow the predecessor's + /// `next` onto it and stop, so a later rewrite of that predecessor is + /// not an interior-link break. + fn retire_head_page( + &self, + body: &TableFile, + alloc: &mut AllocState, + scripthash: &[u8; 32], + home: LocatedHome, + head_page: u64, + pack: &BoundaryPack, + ) -> Result<(), StoreError> { + let mut page = [0u8; SH_PAGE_SIZE]; + sh_page_pack_extent_last_fks(&mut page, &[], pack.base, pack.extent_n, 0)?; + sh_page_set_extent_creates(&mut page, pack.creates); + self.commit_tail_page( + body, + alloc, + scripthash, + home, + TailWrite { + target: head_page, + bytes: &page, + new_head: None, + head_page, + removed: 0, + }, + ) + } + + fn rollback_undo( + &self, + body: &TableFile, + alloc: &mut AllocState, + rec: &UndoRec, + ) -> Result<(), StoreError> { + restore_page(body, rec)?; + if let Some(word) = &rec.word { + self.restore_head_word(word)?; + } + clear_undo_keep_image(body, alloc)?; + self.free_slab(body, alloc, SH_PAGE_SLAB_CLASS, rec.image)?; + Ok(()) + } + + fn recover_one_undo( + &self, + body: &TableFile, + alloc_mu: &std::sync::Mutex, + ) -> Result<(), StoreError> { + let Some(rec) = read_undo(body)? else { + return Ok(()); + }; + let mut alloc = alloc_mu.lock().unwrap(); + self.rollback_undo(body, &mut alloc, &rec)?; + write_alloc_header(body, &alloc)?; + body.sync_data_only()?; + Ok(()) + } + + fn capture_head_word( + &self, + scripthash: &[u8; 32], + home: LocatedHome, + ) -> Result, StoreError> { + let hk = super::head_key_from_full(scripthash); + match home { + LocatedHome::Main => { + let si = self.shard_index(scripthash); + if let Some(slot) = self.sorted_main.get(si) { + let g = slot.read().unwrap(); + if let Some(h) = g.as_ref() { + if let Some(off) = h.value_word_offset(&hk)? { + let old = h.read_value_word(off)?; + return Ok(Some(HeadWord { + tag: HEAD_MAIN, + index: si as u16, + off, + old, + })); + } + } + } + self.capture_ingest_word(scripthash) + } + LocatedHome::Ingest => self.capture_ingest_word(scripthash), + LocatedHome::SealedOvf => { + let g = self.sealed_ovf.lock().unwrap(); + for (i, h) in g.iter().enumerate().rev() { + if let Some(off) = h.value_word_offset(&hk)? { + let old = h.read_value_word(off)?; + return Ok(Some(HeadWord { + tag: HEAD_SORTED_OVF, + index: i as u16, + off, + old, + })); + } + } + drop(g); + if let Some(l1) = self.ovf_l1.lock().unwrap().as_ref() { + if let Some(off) = l1.head.value_word_offset(&hk)? { + let old = l1.head.read_value_word(off)?; + return Ok(Some(HeadWord { + tag: HEAD_OVF_L1, + index: 0, + off, + old, + })); + } + } + Ok(None) + } + } + } + + fn capture_ingest_word(&self, scripthash: &[u8; 32]) -> Result, StoreError> { + let ingest = self.ingest.lock().unwrap(); + let Some(off) = ingest.value_word_offset(scripthash)? else { + return Ok(None); + }; + let old = ingest.read_value_word(off)?; + Ok(Some(HeadWord { + tag: HEAD_INGEST, + index: 0, + off, + old, + })) + } + + fn restore_head_word(&self, word: &HeadWord) -> Result<(), StoreError> { + match word.tag { + HEAD_NONE => Ok(()), + HEAD_MAIN => { + let si = usize::from(word.index); + let Some(slot) = self.sorted_main.get(si) else { + return Err(StoreError::Corrupt("scripthash undo: main shard missing")); + }; + let g = slot.read().unwrap(); + let Some(h) = g.as_ref() else { + return Err(StoreError::Corrupt("scripthash undo: main head missing")); + }; + h.write_value_word(word.off, &word.old)?; + h.flush() + } + HEAD_INGEST => { + let ingest = self.ingest.lock().unwrap(); + ingest.write_value_word(word.off, &word.old)?; + ingest.flush() + } + HEAD_SORTED_OVF => { + let g = self.sealed_ovf.lock().unwrap(); + let Some(h) = g.get(usize::from(word.index)) else { + return Err(StoreError::Corrupt("scripthash undo: ovf head missing")); + }; + h.write_value_word(word.off, &word.old)?; + h.flush() + } + HEAD_OVF_L1 => { + let g = self.ovf_l1.lock().unwrap(); + let Some(l1) = g.as_ref() else { + return Err(StoreError::Corrupt("scripthash undo: ovf l1 missing")); + }; + l1.head.write_value_word(word.off, &word.old)?; + l1.head.flush() + } + _ => Err(StoreError::Corrupt("scripthash undo: bad head tag")), + } + } + + fn sync_home(&self, home: LocatedHome, scripthash: &[u8; 32]) -> Result<(), StoreError> { + match home { + LocatedHome::Main => { + let si = self.shard_index(scripthash); + if let Some(slot) = self.sorted_main.get(si) { + if let Some(h) = slot.read().unwrap().as_ref() { + h.flush()?; + } + } + self.ingest.lock().unwrap().flush() + } + LocatedHome::Ingest => self.ingest.lock().unwrap().flush(), + LocatedHome::SealedOvf => { + for h in self.sealed_ovf.lock().unwrap().iter() { + h.flush()?; + } + if let Some(l1) = self.ovf_l1.lock().unwrap().as_ref() { + l1.head.flush()?; + } + Ok(()) + } + } + } +} + +struct PreparedTail { + at: u64, + bytes: [u8; SH_PAGE_SIZE], + new_last: u64, +} + +struct BoundaryPack { + base: u64, + extent_n: u32, + boundary: u64, + creates: u32, +} + +struct TailWrite<'a> { + target: u64, + bytes: &'a [u8; SH_PAGE_SIZE], + new_head: Option, + /// Page the head names while `target` is written. + head_page: u64, + removed: u64, +} + +fn prepare_boundary( + table: &ScriptHashTable, + body: &TableFile, + alloc: &mut AllocState, + pack: &BoundaryPack, + keep: &[u64], +) -> Result { + if keep.is_empty() { + return Err(StoreError::Corrupt("scripthash suffix keep is empty")); + } + let fks: Vec = keep.iter().copied().map(Fk).collect(); + let chunks = sh_page_chunk_ranges(&fks)?; + if chunks.len() > 2 { + return Err(StoreError::Corrupt( + "scripthash suffix keep exceeds two pages", + )); + } + if chunks.len() == 1 { + let mut page = [0u8; SH_PAGE_SIZE]; + sh_page_pack_extent_last_fks(&mut page, keep, pack.base, pack.extent_n, 0)?; + sh_page_set_extent_creates(&mut page, pack.creates); + return Ok(PreparedTail { + at: pack.boundary, + bytes: page, + new_last: pack.boundary, + }); + } + let (s0, e0) = chunks[0]; + let (s1, e1) = chunks[1]; + let tail_off = table.alloc_page(body, alloc)?; + let mut tail = [0u8; SH_PAGE_SIZE]; + sh_page_pack_extent_last_fks(&mut tail, &keep[s1..e1], pack.base, pack.extent_n, 0)?; + sh_page_set_extent_creates(&mut tail, pack.creates); + body.write_at(tail_off, &tail)?; + let mut mid = [0u8; SH_PAGE_SIZE]; + sh_page_pack_fks(&mut mid, &keep[s0..e0], tail_off)?; + Ok(PreparedTail { + at: pack.boundary, + bytes: mid, + new_last: tail_off, + }) +} + +fn new_extent_n(base: u64, old_n: u32, boundary: u64) -> u32 { + if old_n == 0 + || base == 0 + || boundary < base + || !(boundary - base).is_multiple_of(SH_PAGE_SIZE as u64) + { + return old_n; + } + let idx = (boundary - base) / SH_PAGE_SIZE as u64; + let end = + base.saturating_add(u64::from(old_n.saturating_sub(1)).saturating_mul(SH_PAGE_SIZE as u64)); + if boundary <= end && idx < u64::from(u32::MAX) { + (idx as u32).saturating_add(1) + } else { + old_n + } +} + +fn new_creates(old: u32, removed: u64) -> u32 { + if old == 0 { + 0 + } else { + old.saturating_sub(u32::try_from(removed).unwrap_or(u32::MAX)) + } +} + +fn plan_extent_suffix( + body: &TableFile, + last_page: u64, + remove: &BTreeSet, +) -> Result { + if remove.is_empty() { + return Ok(SuffixPlan::Absent); + } + let last = read_page(body, last_page)?; + let Some((base, old_n)) = sh_page_extent(sh_page_as_array(&last)?)? else { + return Err(StoreError::Corrupt("scripthash extent last page not ver=2")); + }; + let creates = sh_page_extent_creates(sh_page_as_array(&last)?); + let mut off = last_page; + let mut dropped_pages = Vec::new(); + let mut found = BTreeSet::new(); + let cap = remove.len().saturating_add(2); + for _ in 0..cap { + let page = read_page(body, off)?; + let fks = page_fks(&page)?; + if fks.is_empty() { + let Some(pred) = predecessor(body, base, off)? else { + return if found.is_empty() { + Ok(SuffixPlan::Absent) + } else { + Ok(SuffixPlan::DropAll { + removed: found.len() as u64, + }) + }; + }; + if pred == off { + return Err(StoreError::Corrupt("scripthash extent empty tail loops")); + } + dropped_pages.push(off); + off = pred; + continue; + } + let first = fks[0].0; + let mut keep = Vec::new(); + let mut dropped_here = Vec::new(); + for fk in &fks { + if remove.contains(&fk.0) { + dropped_here.push(fk.0); + } else { + keep.push(fk.0); + } + } + if let Some(max_keep) = keep.iter().copied().max() { + if dropped_here.iter().any(|&d| d < max_keep) { + return Ok(SuffixPlan::NotSuffix); + } + } + found.extend(dropped_here.iter().copied()); + if !keep.is_empty() { + if remove.iter().any(|r| *r < first && !found.contains(r)) { + return Ok(SuffixPlan::NotSuffix); + } + if found.is_empty() { + return Ok(SuffixPlan::Absent); + } + return Ok(SuffixPlan::Rewrite { + boundary: off, + keep, + drop_pages: dropped_pages, + base, + old_n, + creates, + removed: found.len() as u64, + }); + } + dropped_pages.push(off); + let more_below = remove.iter().any(|r| *r < first && !found.contains(r)); + if !more_below { + return finish_dropped_page( + body, + base, + old_n, + creates, + off, + dropped_pages, + found.len() as u64, + ); + } + let Some(pred) = predecessor(body, base, off)? else { + return if found.is_empty() { + Ok(SuffixPlan::Absent) + } else { + Ok(SuffixPlan::DropAll { + removed: found.len() as u64, + }) + }; + }; + off = pred; + } + Ok(SuffixPlan::NotSuffix) +} + +fn finish_dropped_page( + body: &TableFile, + base: u64, + old_n: u32, + creates: u32, + off: u64, + dropped_pages: Vec, + removed: u64, +) -> Result { + let Some(pred) = predecessor(body, base, off)? else { + return Ok(SuffixPlan::DropAll { removed }); + }; + let pred_page = read_page(body, pred)?; + let pred_fks = page_fks(&pred_page)?; + if pred_fks.is_empty() { + return Err(StoreError::Corrupt( + "scripthash extent predecessor page is empty", + )); + } + Ok(SuffixPlan::Rewrite { + boundary: pred, + keep: pred_fks.iter().map(|fk| fk.0).collect(), + drop_pages: dropped_pages, + base, + old_n, + creates, + removed, + }) +} + +fn predecessor(body: &TableFile, base: u64, page: u64) -> Result, StoreError> { + if page >= base.saturating_add(SH_PAGE_SIZE as u64) + && (page - base).is_multiple_of(SH_PAGE_SIZE as u64) + { + let pred = page - SH_PAGE_SIZE as u64; + if pred >= base { + let prev = read_page(body, pred)?; + if sh_page_next(sh_page_as_array(&prev)?)? == page { + return Ok(Some(pred)); + } + } + } + let mut prev = None; + let mut cur = base; + if cur == 0 { + return Ok(None); + } + for _ in 0..100_000 { + if cur == page { + return Ok(prev); + } + let page_bytes = read_page(body, cur)?; + let next = sh_page_next(sh_page_as_array(&page_bytes)?)?; + if next == 0 { + return Ok(None); + } + prev = Some(cur); + cur = next; + } + Err(StoreError::Corrupt( + "scripthash extent predecessor walk ran away", + )) +} + +fn read_page(body: &TableFile, off: u64) -> Result<[u8; SH_PAGE_SIZE], StoreError> { + let mut page = [0u8; SH_PAGE_SIZE]; + body.read_at(off, &mut page)?; + Ok(page) +} + +fn page_fks(page: &[u8; SH_PAGE_SIZE]) -> Result, StoreError> { + let mut out = Vec::new(); + sh_page_entries_into(page, &mut out)?; + Ok(out) +} + +fn arm_undo( + table: &ScriptHashTable, + body: &TableFile, + alloc: &mut AllocState, + target: u64, + word: Option, +) -> Result { + let old_page = read_page(body, target)?; + let image = table.alloc_page(body, alloc)?; + body.write_at(image, &old_page)?; + // The image and the header are different blocks. Publish SHUN only after + // the image is durable, or recovery copies an unwritten page onto the tail. + body.sync_data_only()?; + let rec = UndoRec { + target, + image, + word, + }; + persist_undo(body, alloc, Some(&rec))?; + body.sync_data_only()?; + Ok(rec) +} + +fn restore_page(body: &TableFile, rec: &UndoRec) -> Result<(), StoreError> { + let image = read_page(body, rec.image)?; + body.write_at(rec.target, &image)?; + body.sync_data_only()?; + Ok(()) +} + +fn clear_undo_keep_image(body: &TableFile, alloc: &AllocState) -> Result<(), StoreError> { + #[cfg(test)] + if FAIL_CLEAR_UNDO.with(|c| c.replace(false)) { + return Err(StoreError::Corrupt("scripthash undo clear failed")); + } + persist_undo(body, alloc, None)?; + body.sync_data_only()?; + Ok(()) +} + +fn frees_old_pages(old: &ShHeadValue) -> bool { + matches!(old, ShHeadValue::Extent { .. } | ShHeadValue::Slab { .. }) +} + +fn note_batch_body( + table: &ScriptHashTable, + batch: &mut UnlinkBatch, + scripthash: &[u8; 32], + home: LocatedHome, +) { + match home { + LocatedHome::Main => { + let si = table.shard_index(scripthash); + if !batch.main_shards.contains(&si) { + batch.main_shards.push(si); + } + } + LocatedHome::Ingest | LocatedHome::SealedOvf => batch.ovf = true, + } +} + +fn note_readable(body: &TableFile, last_page: u64) { + #[cfg(test)] + if WATCH_READABLE.with(|c| c.get()) && super::collect_extent_then_tail(body, last_page).is_err() + { + READABLE_BROKE.with(|c| c.set(true)); + } + #[cfg(not(test))] + { + let _ = (body, last_page); + } +} + +fn persist_undo( + body: &TableFile, + alloc: &AllocState, + undo: Option<&UndoRec>, +) -> Result<(), StoreError> { + let mut buf = vec![0u8; UNDO_AT + UNDO_LEN]; + fill_alloc_prefix(&mut buf, alloc); + if let Some(rec) = undo { + encode_undo(&mut buf[UNDO_AT..], rec)?; + } + body.write_at(crate::file::FILE_HEADER_LEN as u64, &buf) +} + +fn fill_alloc_prefix(buf: &mut [u8], state: &AllocState) { + buf[0..4].copy_from_slice(&super::SH_ALLOC_MAGIC); + buf[4..6].copy_from_slice(&super::SH_ALLOC_VERSION.to_le_bytes()); + buf[8..16].copy_from_slice(&state.live_count.to_le_bytes()); + buf[16..24].copy_from_slice(&state.bump.to_le_bytes()); + let mut off = 24usize; + for h in &state.free_head { + if off + 8 > UNDO_AT { + break; + } + buf[off..off + 8].copy_from_slice(&h.to_le_bytes()); + off += 8; + } +} + +fn encode_undo(buf: &mut [u8], rec: &UndoRec) -> Result<(), StoreError> { + if buf.len() < UNDO_LEN { + return Err(StoreError::Corrupt("scripthash undo record short")); + } + buf[..4].copy_from_slice(&UNDO_MAGIC); + let word = rec.word.as_ref(); + buf[4] = word.map(|w| w.tag).unwrap_or(HEAD_NONE); + buf[6..8].copy_from_slice(&word.map(|w| w.index).unwrap_or(0).to_le_bytes()); + buf[8..16].copy_from_slice(&rec.target.to_le_bytes()); + buf[16..24].copy_from_slice(&rec.image.to_le_bytes()); + buf[24..32].copy_from_slice(&word.map(|w| w.off).unwrap_or(0).to_le_bytes()); + buf[32..40].copy_from_slice(word.map(|w| &w.old).unwrap_or(&[0u8; 8])); + Ok(()) +} + +fn read_undo(body: &TableFile) -> Result, StoreError> { + let mut buf = [0u8; UNDO_LEN]; + let at = crate::file::FILE_HEADER_LEN as u64 + UNDO_AT as u64; + if body.logical_len() < at + UNDO_LEN as u64 { + return Ok(None); + } + body.read_at(at, &mut buf)?; + if buf[..4] != UNDO_MAGIC { + return Ok(None); + } + let tag = buf[4]; + let index = u16::from_le_bytes(buf[6..8].try_into().unwrap()); + let target = u64::from_le_bytes(buf[8..16].try_into().unwrap()); + let image = u64::from_le_bytes(buf[16..24].try_into().unwrap()); + let head_off = u64::from_le_bytes(buf[24..32].try_into().unwrap()); + let mut old = [0u8; 8]; + old.copy_from_slice(&buf[32..40]); + if target == 0 + || image == 0 + || !target.is_multiple_of(SH_PAGE_SIZE as u64) + || !image.is_multiple_of(SH_PAGE_SIZE as u64) + || image >= body.logical_len() + || target >= body.logical_len() + { + return Err(StoreError::Corrupt("scripthash undo record is not a page")); + } + let word = if tag == HEAD_NONE { + None + } else { + Some(HeadWord { + tag, + index, + off: head_off, + old, + }) + }; + Ok(Some(UndoRec { + target, + image, + word, + })) +} diff --git a/crates/rbitcoin-test/tests/scenarios.rs b/crates/rbitcoin-test/tests/scenarios.rs index 6b6361d2c5..5a4960f850 100644 --- a/crates/rbitcoin-test/tests/scenarios.rs +++ b/crates/rbitcoin-test/tests/scenarios.rs @@ -832,6 +832,17 @@ fn pin_disconnect_to_genesis_reconnect_and_tip_shrink( "poison at n-3 shrinks to last good n-4" ); pin_megakey_block_disconnect(&q, tip, hashes[tip.0 as usize]); + let sh = rbitcoin_store::script_hash(&[0x99]); + let store = td.store_path(); + let expect_tip = q.tip_height(); + q.flush().unwrap(); + drop(q); + let q = Query::open_or_create_tiny(&store).unwrap(); + assert_eq!(q.tip_height(), expect_tip, "tip changed on reopen"); + assert!( + q.scripthash_history(&sh).unwrap().is_empty(), + "reopen keeps the disconnected megakey gone" + ); } /// Disconnect genesis too, then rebuild from an empty chain with a @@ -924,7 +935,8 @@ fn pin_disconnect_logs_and_rewinds_ibd_marks(q: &Query, top: u32, top_hash: &[u8 /// 257 creates of one script in one block make its scripthash head an /// extent. Disconnecting that block unlinks every create and truncates the -/// tweak index with it. +/// tweak index with it. A multi-page suffix is a store test: building it +/// here means archiving thousands of transactions. fn pin_megakey_block_disconnect(q: &Query, tip: Height, tip_hash: [u8; 32]) { use rbitcoin_query::TxApply; use rbitcoin_store::{script_hash, InputRecord, OutputRecord, ShHeadValue}; diff --git a/docs/crash-recovery.md b/docs/crash-recovery.md index 19a25bb0f4..07b546caba 100644 --- a/docs/crash-recovery.md +++ b/docs/crash-recovery.md @@ -79,8 +79,14 @@ Open revalidation runs in `Query::open_or_create` **before** P2P can extend tip. `scripthash.body` once into a process set of `create_tx_fk`s already present; re-confirm skips those txs. Hot path only appends + maintains heads in RAM. - Creates for unstrong / above-tip txs are **invisible** via `is_confirmed_strong`. -- Disconnect tip: **unlink** creates for that block’s outputs (tombstone + rewire); - process set updated so re-confirm can re-index. +- Disconnect tip: **one unlink per scripthash**. A megakey drops that block's + creates from the tail pages. The saved page image is synced before the undo + record is published. The page the head names is updated before any earlier + page is rewritten, then the head moves. Old pages are freed only after that. + If clearing the undo fails, the saved page and the old head word are written + back. A crash does the same restore, so the same disconnect can run again. + A non-suffix unlink writes the replacement, syncs those bytes and the new + heads once for the whole disconnect, then frees the old pages. - No tip-mode full rebuild; corrupt index ⇒ reindex (wipe store / redo IBD). ## Flush