use std::collections::{BTreeMap, HashSet};
use std::fs::{File, OpenOptions};
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use anyhow::{Context, Result};
use hashtree_core::{Cid, HashTree, HashTreeConfig, Store};
use hashtree_index::{BTree, BTreeOptions};
use hashtree_lmdb::{ReadOnlyPoolStore, SHARED_BLOB_POOL_DIR_NAME};
use hashtree_nostr::{
nostr_event_index_entries, ListEventsOptions, NostrEventIndex, NostrEventStore,
StoredNostrEvent,
};
use heed::types::Bytes;
use heed::{Database, EnvFlags, EnvOpenOptions};
use nostr::{Event, JsonUtil};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use super::super::{
cid_to_nhash, parse_root_text, StagedNostrCrawlState, STAGE_DIR, STAGE_FORMAT_VERSION,
STAGE_STATE_FILE,
};
use super::tranche::load_v3_candidate_audit_authority;
#[cfg(test)]
use super::tranche::V3CandidateAuditAuthority;
use super::{
bulk_paths, encode_cid, validate_terminal_stage_state, BulkProjectionSpool,
BulkProjectionState, EntryTrieCursor, SpoolEventRecord, BULK_PROJECTION_VERSION,
};
mod stage_corpus;
use stage_corpus::{reconcile_v3_stage_corpus, BulkProjectionStageCorpusAudit, V3StageCorpusSpec};
const PROFILE_RANK_DECISION_FORMAT: &str = "iris-social/profile-search-v3-rank-decisions@1";
const PROFILE_RANK_DECISION_REPORT_FORMAT: &str =
"iris-social/profile-search-v3-rank-decision-artifacts@1";
const PROFILE_RANK_DECISION_CENSUS_FORMAT: &str = "iris-social/social-graph-crawl-census@2";
const PROFILE_RANK_POLICY: &str = "follow-distance@1";
const PROFILE_EXCLUSION_POLICY: &str = "all-nonselected-graph-identities@1";
const PROFILE_OVERMUTE_THRESHOLD: usize = 1;
const PROFILE_UNREACHABLE_DISTANCE: u32 = 1_000;
const BULK_PROJECTION_AUDIT_FORMAT_VERSION: u32 = 4;
const DEFAULT_AUDIT_BTREE_ORDER: usize = 64;
const FULL_POLICY_CUTOVER_AUDIT_MODE: &str = "full-policy-cutover";
const RECOVERY_TRANCHE_AUDIT_MODE: &str = "recovery-tranche-internal-non-cutover";
const EXACT_INDEX_PARITY_FORMAT: &str = "hashtree/nostr-bulk-projection-exact-index-parity@1";
#[derive(Debug, Clone)]
pub(crate) struct BulkProjectionAuditOptions {
pub(crate) staging_data_dir: PathBuf,
pub(crate) expected_state_sha256: String,
pub(crate) v3_candidate: bool,
pub(crate) expected_stage_state_sha256: Option<String>,
pub(crate) expected_policy_sha256: Option<String>,
pub(crate) expected_profile_distance_seal_sha256: Option<String>,
pub(crate) profile_rank_decisions_file: Option<PathBuf>,
pub(crate) expected_profile_rank_decisions_file_sha256: Option<String>,
pub(crate) profile_rank_decisions_report: Option<PathBuf>,
pub(crate) expected_profile_rank_decisions_report_sha256: Option<String>,
pub(crate) expected_full_author_count: Option<usize>,
pub(crate) allow_recovery_tranche: bool,
pub(crate) btree_order: Option<usize>,
pub(crate) page_size: usize,
pub(crate) query_limit: usize,
pub(crate) out: PathBuf,
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "lowercase")]
pub(super) enum AuditSubjectKind {
V2,
V3,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum AuditScope {
RecoveryTranche,
FullPolicyCutover,
}
impl AuditScope {
fn audit_mode(self) -> &'static str {
match self {
Self::RecoveryTranche => RECOVERY_TRANCHE_AUDIT_MODE,
Self::FullPolicyCutover => FULL_POLICY_CUTOVER_AUDIT_MODE,
}
}
fn recovery_tranche_only(self) -> bool {
self == Self::RecoveryTranche
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct PinnedAuditSnapshot {
label: String,
path: PathBuf,
bytes: Vec<u8>,
sha256: String,
}
impl PinnedAuditSnapshot {
fn read(label: impl Into<String>, path: &Path, expected_sha256: &str) -> Result<Self> {
let label = label.into();
let path = path
.canonicalize()
.with_context(|| format!("canonicalize {label} {}", path.display()))?;
if !path.is_file() {
anyhow::bail!("{label} is not a regular file: {}", path.display());
}
let bytes =
std::fs::read(&path).with_context(|| format!("read {label} {}", path.display()))?;
let sha256 = bytes_sha256(&bytes);
require_expected_sha256(&label, &sha256, expected_sha256)?;
Ok(Self {
label,
path,
bytes,
sha256,
})
}
fn recheck(&self) -> Result<()> {
let current = std::fs::read(&self.path)
.with_context(|| format!("re-read {} {}", self.label, self.path.display()))?;
if current != self.bytes {
anyhow::bail!("{} changed during read-only audit", self.label);
}
Ok(())
}
}
#[derive(Debug, Clone, PartialEq)]
pub(super) struct AuditSubject {
kind: AuditSubjectKind,
version: u32,
candidate_root: Cid,
built_roots: BTreeMap<u8, String>,
state_sha256: String,
stage_state_sha256: String,
trusted_policy_sha256: String,
policy_author_allowlist_sha256: String,
expected_profile_distance_seal_sha256: Option<String>,
expected_profile_distance_provenance: Option<ProfileDistanceProvenance>,
trusted_full_author_count: usize,
crawl_policy_max_follow_distance: Option<u32>,
authors_processed: usize,
authors_total: usize,
btree_order: usize,
scope: AuditScope,
spool_path: PathBuf,
snapshots: Vec<PinnedAuditSnapshot>,
v3_recheck: Option<V3AuditSubjectSpec>,
stage_corpus: Option<V3StageCorpusSpec>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[allow(dead_code)]
pub(super) struct V3AuditSubjectSpec {
pub(super) expected_state_sha256: String,
pub(super) staging_data_dir: PathBuf,
}
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
struct BulkProjectionIndexAudit {
index: String,
root: Option<String>,
nodes: u64,
links: u64,
durable_values_validated: u64,
entries_sha256: String,
retained_set_sha256: String,
first_key: Option<String>,
last_key: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub(super) struct BulkProjectionExactIndexParityRoot {
pub(super) index: String,
pub(super) stable_id: u8,
pub(super) root: String,
pub(super) nodes: u64,
pub(super) entries: u64,
pub(super) durable_values_validated: u64,
pub(super) ordered_entries_sha256: String,
pub(super) retained_set_sha256: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub(super) struct BulkProjectionExactIndexParityEvidence {
pub(super) format: String,
pub(super) btree_order: usize,
pub(super) spool_event_records: u64,
pub(super) indexes: Vec<BulkProjectionExactIndexParityRoot>,
pub(super) evidence_sha256: String,
}
#[derive(Debug, Clone, PartialEq)]
pub(super) struct BulkProjectionMissingEventBlob {
pub(super) event_id: String,
pub(super) cid: Cid,
}
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
struct BulkProjectionQueryAudit {
query: String,
parameters: serde_json::Value,
event_ids: Vec<String>,
}
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
struct BulkProjectionBlockEvidence {
role: String,
nhash: String,
sha256: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub(super) struct BulkProjectionProfileAudit {
pub(super) by_pubkey_root: String,
pub(super) by_pubkey_root_file_sha256: String,
pub(super) by_pubkey_nodes: u64,
pub(super) by_pubkey_links: u64,
pub(super) by_pubkey_entries_sha256: String,
pub(super) search_root: String,
pub(super) search_root_file_sha256: String,
pub(super) search_nodes: u64,
pub(super) search_entries: u64,
pub(super) search_entries_sha256: String,
pub(super) sample_pubkey: String,
pub(super) sample_event_id: String,
pub(super) sample_name: String,
pub(super) follow_distance_binding: String,
pub(super) follow_distance_seal_sha256: String,
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct StrictStoredProfileSearchEntry {
pubkey: String,
name: String,
aliases: Vec<String>,
nip05: Option<String>,
#[serde(default)]
follow_distance: Option<u32>,
created_at: u64,
event_nhash: String,
}
impl From<StrictStoredProfileSearchEntry> for hashtree_cli::socialgraph::StoredProfileSearchEntry {
fn from(value: StrictStoredProfileSearchEntry) -> Self {
Self {
pubkey: value.pubkey,
name: value.name,
aliases: value.aliases,
nip05: value.nip05,
follow_distance: value.follow_distance,
created_at: value.created_at,
event_nhash: value.event_nhash,
}
}
}
#[derive(Debug, Clone)]
struct ExpectedProfileSearchEntry {
pubkey: String,
event: Event,
mirrored_cid: Cid,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub(super) struct ProfileDistanceProvenance {
pub(super) format: String,
pub(super) census_format: String,
pub(super) rank_decisions_file_sha256: String,
pub(super) rank_decisions_report_sha256: String,
pub(super) rank_decisions_sha256: String,
pub(super) social_graph_root: String,
pub(super) social_graph_sha256: String,
pub(super) eligible_authors_sha256: String,
pub(super) record_count: usize,
pub(super) eligible_count: usize,
pub(super) excluded_count: usize,
pub(super) overmute_threshold: usize,
pub(super) census_max_distance: Option<u32>,
pub(super) rank_policy: String,
pub(super) exclusion_policy: String,
}
#[derive(Debug)]
pub(super) struct TrustedProfileRankDecisions {
pub(super) decisions: BTreeMap<String, Option<u32>>,
pub(super) decisions_path: PathBuf,
pub(super) decisions_bytes: Vec<u8>,
pub(super) report_path: PathBuf,
pub(super) report_bytes: Vec<u8>,
pub(super) evidence: ProfileDistanceProvenance,
}
#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct ProfileRankDecisionHeader {
format: String,
eligible_ranks_sha256: String,
record_count: usize,
}
#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct ProfileRankDecisionRecord {
pubkey: String,
decision: String,
rank_hint: Option<u32>,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct ProfileRankDecisionReport {
format: String,
census_format: String,
social_graph_root: String,
social_graph_sha256: String,
eligible_authors_sha256: String,
overmute_threshold: usize,
max_distance: Option<u32>,
rank_policy: String,
exclusion_policy: String,
record_count: usize,
eligible_count: usize,
excluded_count: usize,
reachable_count: usize,
reachable_overmuted_count: usize,
distance_excluded_count: usize,
unreachable_count: usize,
all_graph_overmuted_count: usize,
rank_decisions_sha256: String,
rank_decisions_file_sha256: String,
}
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
pub(crate) struct BulkProjectionAuditOutput {
version: u32,
subject_kind: AuditSubjectKind,
subject_version: u32,
candidate_root: String,
state_sha256: String,
stage_state_sha256: String,
trusted_policy_sha256: String,
policy_author_allowlist_sha256: String,
trusted_profile_distance_seal_sha256: Option<String>,
profile_distance_provenance: Option<ProfileDistanceProvenance>,
trusted_full_author_count: usize,
crawl_policy_max_follow_distance: Option<u32>,
audit_mode: String,
cutover_eligible: bool,
pool_catalog_sha256: String,
pool_manifest_sha256: String,
pool_stored_locations: u64,
authors_processed: usize,
authors_total: usize,
recovery_tranche_only: bool,
stage_corpus: Option<BulkProjectionStageCorpusAudit>,
indexes: Vec<BulkProjectionIndexAudit>,
profile: BulkProjectionProfileAudit,
queries: Vec<BulkProjectionQueryAudit>,
representative_blocks: Vec<BulkProjectionBlockEvidence>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[allow(dead_code)]
pub(super) struct FullCutoverVerified {
evidence_sha256: String,
subject_kind: AuditSubjectKind,
subject_version: u32,
candidate_root: String,
subject_state_sha256: String,
stage_state_sha256: String,
trusted_policy_sha256: String,
policy_author_allowlist_sha256: String,
trusted_full_author_count: usize,
crawl_policy_max_follow_distance: Option<u32>,
authors_processed: usize,
authors_total: usize,
index_roots: BTreeMap<u8, String>,
pool_catalog_sha256: String,
pool_manifest_sha256: String,
pool_stored_locations: u64,
stage_segment_count: u64,
stage_event_cids_validated: u64,
stage_retained_records: u64,
stage_retained_records_sha256: String,
stage_replaceable_slots: u64,
stage_replaceable_slots_sha256: String,
profile_by_pubkey_root: String,
profile_search_root: String,
profile_follow_distance_seal_sha256: String,
profile_distance_provenance: ProfileDistanceProvenance,
}
#[derive(Debug)]
pub(super) struct BulkProjectionAuditCompletion {
pub(super) output: BulkProjectionAuditOutput,
pub(super) full_cutover: Option<FullCutoverVerified>,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
struct EntrySetProof {
count: u64,
xor: [u8; 32],
}
impl EntrySetProof {
fn insert(&mut self, index: NostrEventIndex, key: &str, cid: &Cid) -> Result<()> {
let mut digest = Sha256::new();
digest.update(b"hashtree-nostr-retained-index-entry-v1\0");
digest.update([index.stable_id()]);
digest.update((key.len() as u64).to_be_bytes());
digest.update(key.as_bytes());
let encoded_cid = encode_cid(cid);
digest.update((encoded_cid.len() as u64).to_be_bytes());
digest.update(encoded_cid);
let entry: [u8; 32] = digest.finalize().into();
for (target, byte) in self.xor.iter_mut().zip(entry) {
*target ^= byte;
}
self.count = self
.count
.checked_add(1)
.context("retained index entry count overflow")?;
Ok(())
}
fn evidence_sha256(self) -> String {
let mut digest = Sha256::new();
digest.update(b"hashtree-nostr-retained-index-set-v1\0");
digest.update(self.count.to_be_bytes());
digest.update(self.xor);
hex::encode(digest.finalize())
}
}
impl BulkProjectionSpool {
pub(super) fn open_read_only(path: &Path) -> Result<Self> {
if !path.join("data.mdb").is_file() {
anyhow::bail!("bulk projection spool is missing at {}", path.display());
}
let mut options = EnvOpenOptions::new();
options.max_dbs(3).max_readers(32);
unsafe {
options.flags(EnvFlags::READ_ONLY | EnvFlags::NO_READ_AHEAD);
}
let env = unsafe { options.open(path) }
.with_context(|| format!("open bulk projection spool {} read-only", path.display()))?;
let rtxn = env.read_txn()?;
let open_database = |name| -> Result<Database<Bytes, Bytes>> {
env.open_database(&rtxn, Some(name))?
.with_context(|| format!("bulk projection spool omitted {name} database"))
};
let events = open_database("events")?;
let slots = open_database("slots")?;
let entries = open_database("entries")?;
rtxn.commit()?;
Ok(Self {
env,
entries,
events,
slots,
})
}
pub(super) fn event_record(&self, event_id: &str) -> Result<Option<SpoolEventRecord>> {
let rtxn = self.env.read_txn()?;
self.events
.get(&rtxn, event_id.as_bytes())?
.map(|encoded| rmp_serde::from_slice(encoded).context("decode bulk spool event record"))
.transpose()
}
fn event_record_count(&self) -> Result<u64> {
let rtxn = self.env.read_txn()?;
let mut count = 0u64;
for item in self.events.iter(&rtxn)? {
item?;
count = count
.checked_add(1)
.context("bulk spool event record count overflow")?;
}
Ok(count)
}
pub(super) fn retained_profile_records(&self) -> Result<BTreeMap<String, SpoolEventRecord>> {
let rtxn = self.env.read_txn()?;
let mut profiles = BTreeMap::new();
for item in self.events.iter(&rtxn)? {
let (event_id, encoded) = item?;
let event_id =
std::str::from_utf8(event_id).context("bulk spool event key is not UTF-8")?;
let record: SpoolEventRecord =
rmp_serde::from_slice(encoded).context("decode retained profile spool event")?;
if record.event.id != event_id {
anyhow::bail!(
"bulk spool event key `{event_id}` differs from record id `{}`",
record.event.id
);
}
if record.event.kind != 0 {
continue;
}
if profiles
.insert(record.event.pubkey.clone(), record)
.is_some()
{
anyhow::bail!("bulk spool retained multiple kind-0 winners for one pubkey");
}
}
Ok(profiles)
}
}
fn bytes_sha256(bytes: &[u8]) -> String {
let mut digest = Sha256::new();
digest.update(bytes);
hex::encode(digest.finalize())
}
fn require_expected_sha256(label: &str, actual: &str, expected: &str) -> Result<()> {
if expected.len() != 64
|| !expected
.bytes()
.all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
{
anyhow::bail!("{label} SHA-256 pin must be 64 lowercase hexadecimal characters");
}
if expected != actual {
anyhow::bail!("{label} SHA-256 mismatch: expected {expected}, found {actual}");
}
Ok(())
}
fn require_sha256_text(label: &str, value: &str) -> Result<()> {
require_expected_sha256(label, value, value)
}
fn validate_complete_built_roots(label: &str, built_roots: &BTreeMap<u8, String>) -> Result<()> {
if built_roots.len() != NostrEventIndex::ALL.len()
|| NostrEventIndex::ALL
.iter()
.any(|index| !built_roots.contains_key(&index.stable_id()))
{
anyhow::bail!("{label} must contain exactly all nine index roots");
}
for index in NostrEventIndex::ALL {
let encoded = built_roots
.get(&index.stable_id())
.expect("all nine root keys checked");
if encoded.is_empty() {
anyhow::bail!("{label} has an empty canonical `{}` root", index.name());
}
let root = parse_root_text(encoded)
.with_context(|| format!("parse {label} {} root", index.name()))?;
if cid_to_nhash(&root)? != *encoded {
anyhow::bail!("{label} {} root is not canonical nhash text", index.name());
}
}
Ok(())
}
fn load_v2_audit_subject(
data_dir: &Path,
options: &BulkProjectionAuditOptions,
) -> Result<AuditSubject> {
let expected_stage_state_sha256 = options
.expected_stage_state_sha256
.as_deref()
.context("v2 audit requires the expected staging-state SHA-256")?;
let expected_policy_sha256 = options
.expected_policy_sha256
.as_deref()
.context("v2 audit requires the expected policy SHA-256")?;
let expected_full_author_count = options
.expected_full_author_count
.context("v2 audit requires the expected full author count")?;
let (state_path, spool_path) = bulk_paths(data_dir);
let stage_state_path = options
.staging_data_dir
.join(STAGE_DIR)
.join(STAGE_STATE_FILE);
let state_snapshot = PinnedAuditSnapshot::read(
"bulk projection state",
&state_path,
&options.expected_state_sha256,
)?;
let stage_snapshot = PinnedAuditSnapshot::read(
"staging state",
&stage_state_path,
expected_stage_state_sha256,
)?;
let state: BulkProjectionState =
serde_json::from_slice(&state_snapshot.bytes).context("parse bulk projection state")?;
let stage: StagedNostrCrawlState =
serde_json::from_slice(&stage_snapshot.bytes).context("parse staging state")?;
if state.version != BULK_PROJECTION_VERSION {
anyhow::bail!(
"unsupported bulk projection state version {}",
state.version
);
}
if stage.version != STAGE_FORMAT_VERSION {
anyhow::bail!("unsupported frozen staging state version {}", stage.version);
}
if state.policy.author_count != expected_full_author_count {
anyhow::bail!(
"bulk projection policy author count mismatch: expected trusted full count {}, found {}",
expected_full_author_count,
state.policy.author_count
);
}
let policy_bytes =
serde_json::to_vec(&state.policy).context("serialize trusted bulk projection policy")?;
let policy_sha256 = bytes_sha256(&policy_bytes);
require_expected_sha256(
"bulk projection policy",
&policy_sha256,
expected_policy_sha256,
)?;
if state.next_author > state.policy.author_count {
anyhow::bail!(
"terminal bulk projection author watermark {} exceeds policy author count {}",
state.next_author,
state.policy.author_count
);
}
if state.policy.max_authors != expected_full_author_count {
anyhow::bail!(
"bulk projection policy max_authors {} differs from trusted full count {}",
state.policy.max_authors,
expected_full_author_count
);
}
let scope = if options.allow_recovery_tranche {
if state.next_author == 0 || state.next_author >= expected_full_author_count {
anyhow::bail!(
"recovery-tranche audit requires a nonzero partial watermark below trusted full \
count {}; found next_author={}",
expected_full_author_count,
state.next_author
);
}
AuditScope::RecoveryTranche
} else {
if state.next_author != expected_full_author_count {
anyhow::bail!(
"full-policy audit requires the terminal and frozen-stage author watermark to \
equal trusted full count {}; found next_author={}; use \
--allow-recovery-tranche only for internal non-cutover evidence",
expected_full_author_count,
state.next_author
);
}
if options.expected_profile_distance_seal_sha256.is_none() {
anyhow::bail!(
"full-policy audit requires the exact independently pinned profile-distance \
seal SHA-256"
);
}
AuditScope::FullPolicyCutover
};
validate_terminal_stage_state(&state, &stage)?;
validate_complete_built_roots("bulk projection state", &state.built_roots)?;
let candidate_root = state
.complete_root
.as_deref()
.context("bulk projection has no complete candidate root")
.and_then(parse_root_text)?;
Ok(AuditSubject {
kind: AuditSubjectKind::V2,
version: state.version,
candidate_root,
built_roots: state.built_roots,
state_sha256: state_snapshot.sha256.clone(),
stage_state_sha256: stage_snapshot.sha256.clone(),
trusted_policy_sha256: policy_sha256,
policy_author_allowlist_sha256: state.policy.author_allowlist_sha256,
expected_profile_distance_seal_sha256: options
.expected_profile_distance_seal_sha256
.clone(),
expected_profile_distance_provenance: None,
trusted_full_author_count: expected_full_author_count,
crawl_policy_max_follow_distance: state.policy.max_follow_distance,
authors_processed: state.next_author,
authors_total: state.policy.author_count,
btree_order: options.btree_order.unwrap_or(DEFAULT_AUDIT_BTREE_ORDER),
scope,
spool_path,
snapshots: vec![state_snapshot, stage_snapshot],
v3_recheck: None,
stage_corpus: None,
})
}
#[cfg_attr(not(test), allow(dead_code))]
pub(super) fn load_v3_audit_subject(
data_dir: &Path,
spec: V3AuditSubjectSpec,
) -> Result<AuditSubject> {
let authority = load_v3_candidate_audit_authority(
data_dir,
&spec.staging_data_dir,
&spec.expected_state_sha256,
)?;
validate_complete_built_roots("v3 audit subject", &authority.built_roots)?;
let candidate_root =
parse_root_text(&authority.candidate_root).context("parse v3 audit candidate root")?;
let state_snapshot = PinnedAuditSnapshot::read(
"v3 tranche state",
&authority.state_path,
&authority.state_sha256,
)?;
let stage_snapshot = PinnedAuditSnapshot::read(
"v3 frozen stage state",
&authority.stage_state_path,
&authority.stage_state_sha256,
)?;
Ok(AuditSubject {
kind: AuditSubjectKind::V3,
version: 3,
candidate_root,
built_roots: authority.built_roots,
state_sha256: state_snapshot.sha256.clone(),
stage_state_sha256: stage_snapshot.sha256.clone(),
trusted_policy_sha256: authority.policy_sha256,
policy_author_allowlist_sha256: authority.policy_author_allowlist_sha256,
expected_profile_distance_seal_sha256: Some(authority.profile_distance_seal_sha256),
expected_profile_distance_provenance: Some(authority.profile_rank_authority),
trusted_full_author_count: authority.author_count,
crawl_policy_max_follow_distance: authority.max_follow_distance,
authors_processed: authority.author_count,
authors_total: authority.author_count,
btree_order: authority.btree_order,
scope: AuditScope::FullPolicyCutover,
spool_path: authority.spool_path,
snapshots: vec![state_snapshot, stage_snapshot],
v3_recheck: Some(spec),
stage_corpus: Some(V3StageCorpusSpec {
staging_data_dir: authority.staging_data_dir,
policy: authority.policy,
expected_segment_count: authority.stage_segment_count,
expected_event_cid_count: authority.stage_event_cid_count,
}),
})
}
#[cfg(test)]
pub(super) async fn reconcile_v3_candidate_stage_corpus_for_test<S: Store>(
authority: &V3CandidateAuditAuthority,
spool: &BulkProjectionSpool,
target: &NostrEventStore<S>,
) -> Result<()> {
let spec = V3StageCorpusSpec {
staging_data_dir: authority.staging_data_dir.clone(),
policy: authority.policy.clone(),
expected_segment_count: authority.stage_segment_count,
expected_event_cid_count: authority.stage_event_cid_count,
};
reconcile_v3_stage_corpus(&spec, spool, target)
.await
.map(|_| ())
}
fn parse_profile_rank_decisions(bytes: &[u8]) -> Result<(BTreeMap<String, Option<u32>>, String)> {
let text = std::str::from_utf8(bytes).context("profile rank-decisions file is not UTF-8")?;
if !text.ends_with('\n') || text.contains('\r') {
anyhow::bail!("profile rank-decisions file must be canonical LF-terminated JSONL");
}
let lines = text[..text.len() - 1].split('\n').collect::<Vec<_>>();
if lines.len() < 2 || lines.iter().any(|line| line.is_empty()) {
anyhow::bail!("profile rank-decisions file requires one header and records");
}
let header: ProfileRankDecisionHeader =
serde_json::from_str(lines[0]).context("strictly decode profile rank-decisions header")?;
if serde_json::to_string(&header)? != lines[0]
|| header.format != PROFILE_RANK_DECISION_FORMAT
|| header.record_count == 0
|| header.record_count != lines.len() - 1
{
anyhow::bail!("profile rank-decisions header is noncanonical or inconsistent");
}
require_sha256_text(
"embedded profile rank-decisions digest",
&header.eligible_ranks_sha256,
)?;
let mut decisions = BTreeMap::new();
let mut previous = None::<String>;
for (position, line) in lines.iter().enumerate().skip(1) {
let record: ProfileRankDecisionRecord = serde_json::from_str(line)
.with_context(|| format!("strictly decode profile rank decision {position}"))?;
if serde_json::to_string(&record)? != *line
|| record.pubkey.len() != 64
|| !record
.pubkey
.bytes()
.all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
{
anyhow::bail!("profile rank decision {position} is noncanonical");
}
if previous
.as_deref()
.is_some_and(|previous| record.pubkey.as_str() <= previous)
{
anyhow::bail!("profile rank decisions are duplicate or not strictly pubkey sorted");
}
let rank = match (record.decision.as_str(), record.rank_hint) {
("eligible", Some(rank)) if rank < PROFILE_UNREACHABLE_DISTANCE => Some(rank),
("excluded", None) => None,
_ => anyhow::bail!("profile rank decision {position} has invalid decision/rank"),
};
previous = Some(record.pubkey.clone());
decisions.insert(record.pubkey, rank);
}
let mut digest = Sha256::new();
digest.update(PROFILE_RANK_DECISION_FORMAT.as_bytes());
digest.update(b"\n");
for (pubkey, rank) in &decisions {
let row = match rank {
Some(rank) => serde_json::json!([pubkey, "eligible", rank]),
None => serde_json::json!([pubkey, "excluded", null]),
};
digest.update(serde_json::to_string(&row)?.as_bytes());
digest.update(b"\n");
}
let decisions_sha256 = hex::encode(digest.finalize());
if decisions_sha256 != header.eligible_ranks_sha256 {
anyhow::bail!(
"profile rank-decisions digest mismatch: embedded {} computed {}",
header.eligible_ranks_sha256,
decisions_sha256
);
}
Ok((decisions, decisions_sha256))
}
fn eligible_profile_rank_authors_sha256(decisions: &BTreeMap<String, Option<u32>>) -> String {
let mut eligible = decisions
.iter()
.filter_map(|(pubkey, rank)| rank.map(|rank| (rank, pubkey)))
.collect::<Vec<_>>();
eligible.sort_unstable();
let mut digest = Sha256::new();
for (_, pubkey) in eligible {
digest.update(pubkey.as_bytes());
digest.update(b"\n");
}
hex::encode(digest.finalize())
}
fn canonical_read_pinned_file(
label: &str,
path: &Path,
expected_sha256: &str,
) -> Result<(PathBuf, Vec<u8>, String)> {
if !path.is_absolute() {
anyhow::bail!("{label} path must be absolute");
}
let path = path
.canonicalize()
.with_context(|| format!("canonicalize {label} {}", path.display()))?;
if !path.is_file() {
anyhow::bail!("{label} is not a regular file: {}", path.display());
}
let bytes = std::fs::read(&path).with_context(|| format!("read {label} {}", path.display()))?;
let sha256 = bytes_sha256(&bytes);
require_expected_sha256(label, &sha256, expected_sha256)?;
Ok((path, bytes, sha256))
}
fn load_trusted_profile_rank_decisions(
options: &BulkProjectionAuditOptions,
) -> Result<Option<TrustedProfileRankDecisions>> {
let supplied = [
options.profile_rank_decisions_file.is_some(),
options
.expected_profile_rank_decisions_file_sha256
.is_some(),
options.profile_rank_decisions_report.is_some(),
options
.expected_profile_rank_decisions_report_sha256
.is_some(),
];
if supplied.iter().all(|supplied| !supplied) {
if options.allow_recovery_tranche {
return Ok(None);
}
anyhow::bail!(
"full-policy cutover requires the independently derived profile rank-decisions \
file/report and both exact SHA-256 pins"
);
}
if supplied.iter().any(|supplied| !supplied) {
anyhow::bail!(
"profile rank-decisions provenance requires file, report, and both exact SHA-256 pins"
);
}
Ok(Some(load_pinned_profile_rank_decisions(
options
.profile_rank_decisions_file
.as_deref()
.expect("all provenance options checked"),
options
.expected_profile_rank_decisions_file_sha256
.as_deref()
.expect("all provenance options checked"),
options
.profile_rank_decisions_report
.as_deref()
.expect("all provenance options checked"),
options
.expected_profile_rank_decisions_report_sha256
.as_deref()
.expect("all provenance options checked"),
)?))
}
pub(super) fn load_pinned_profile_rank_decisions(
profile_rank_decisions_file: &Path,
expected_profile_rank_decisions_file_sha256: &str,
profile_rank_decisions_report: &Path,
expected_profile_rank_decisions_report_sha256: &str,
) -> Result<TrustedProfileRankDecisions> {
let (decisions_path, decisions_bytes, decisions_file_sha256) = canonical_read_pinned_file(
"profile rank-decisions file",
profile_rank_decisions_file,
expected_profile_rank_decisions_file_sha256,
)?;
let (report_path, report_bytes, report_sha256) = canonical_read_pinned_file(
"profile rank-decisions report",
profile_rank_decisions_report,
expected_profile_rank_decisions_report_sha256,
)?;
if decisions_path == report_path {
anyhow::bail!("profile rank-decisions file and report must be distinct");
}
let (decisions, rank_decisions_sha256) = parse_profile_rank_decisions(&decisions_bytes)?;
let report: ProfileRankDecisionReport = serde_json::from_slice(&report_bytes)
.context("strictly decode profile rank-decisions report")?;
for (label, value) in [
(
"report social graph SHA-256",
report.social_graph_sha256.as_str(),
),
(
"report eligible authors SHA-256",
report.eligible_authors_sha256.as_str(),
),
(
"report rank-decisions SHA-256",
report.rank_decisions_sha256.as_str(),
),
(
"report rank-decisions file SHA-256",
report.rank_decisions_file_sha256.as_str(),
),
] {
require_sha256_text(label, value)?;
}
if report.format != PROFILE_RANK_DECISION_REPORT_FORMAT
|| report.census_format != PROFILE_RANK_DECISION_CENSUS_FORMAT
|| report.rank_policy != PROFILE_RANK_POLICY
|| report.exclusion_policy != PROFILE_EXCLUSION_POLICY
|| report.social_graph_root.len() != 64
|| !report
.social_graph_root
.bytes()
.all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
{
anyhow::bail!("profile rank-decisions report has the wrong provenance contract");
}
let eligible_count = decisions.values().filter(|rank| rank.is_some()).count();
let eligible_authors_sha256 = eligible_profile_rank_authors_sha256(&decisions);
let excluded_count = decisions.len() - eligible_count;
let reachable_partition = report
.eligible_count
.checked_add(report.reachable_overmuted_count)
.and_then(|count| count.checked_add(report.distance_excluded_count));
let excluded_partition = report
.reachable_overmuted_count
.checked_add(report.distance_excluded_count)
.and_then(|count| count.checked_add(report.unreachable_count));
if report.record_count != decisions.len()
|| report.eligible_count != eligible_count
|| report.excluded_count != excluded_count
|| report.eligible_count.checked_add(report.excluded_count) != Some(report.record_count)
|| report.reachable_count.checked_add(report.unreachable_count) != Some(report.record_count)
|| reachable_partition != Some(report.reachable_count)
|| excluded_partition != Some(report.excluded_count)
|| report.reachable_overmuted_count > report.reachable_count
|| report.all_graph_overmuted_count > report.record_count
|| report.all_graph_overmuted_count < report.reachable_overmuted_count
|| report.overmute_threshold != PROFILE_OVERMUTE_THRESHOLD
|| report.max_distance.is_some_and(|max_distance| {
decisions
.values()
.flatten()
.any(|rank| *rank > max_distance)
})
|| report.rank_decisions_sha256 != rank_decisions_sha256
|| report.rank_decisions_file_sha256 != decisions_file_sha256
{
anyhow::bail!("profile rank-decisions report does not match its exact decisions artifact");
}
if report.eligible_authors_sha256 != eligible_authors_sha256 {
anyhow::bail!(
"profile rank-decisions report eligible-author digest {} does not match actual \
eligible decisions in census (rank, pubkey) order {}",
report.eligible_authors_sha256,
eligible_authors_sha256
);
}
Ok(TrustedProfileRankDecisions {
decisions,
decisions_path,
decisions_bytes,
report_path,
report_bytes,
evidence: ProfileDistanceProvenance {
format: report.format,
census_format: report.census_format,
rank_decisions_file_sha256: decisions_file_sha256,
rank_decisions_report_sha256: report_sha256,
rank_decisions_sha256,
social_graph_root: report.social_graph_root,
social_graph_sha256: report.social_graph_sha256,
eligible_authors_sha256,
record_count: report.record_count,
eligible_count: report.eligible_count,
excluded_count: report.excluded_count,
overmute_threshold: report.overmute_threshold,
census_max_distance: report.max_distance,
rank_policy: report.rank_policy,
exclusion_policy: report.exclusion_policy,
},
})
}
pub(super) fn recheck_trusted_profile_rank_decisions(
trusted: Option<&TrustedProfileRankDecisions>,
) -> Result<()> {
let Some(trusted) = trusted else {
return Ok(());
};
let decisions = std::fs::read(&trusted.decisions_path).with_context(|| {
format!(
"re-read profile rank-decisions file {}",
trusted.decisions_path.display()
)
})?;
let report = std::fs::read(&trusted.report_path).with_context(|| {
format!(
"re-read profile rank-decisions report {}",
trusted.report_path.display()
)
})?;
if decisions != trusted.decisions_bytes || report != trusted.report_bytes {
anyhow::bail!("profile rank-decisions provenance changed during read-only audit");
}
Ok(())
}
pub(super) fn require_profile_rank_policy_binding(
trusted: Option<&TrustedProfileRankDecisions>,
policy_author_allowlist_sha256: &str,
policy_author_count: usize,
) -> Result<()> {
let Some(trusted) = trusted else {
return Ok(());
};
if trusted.evidence.eligible_authors_sha256 != policy_author_allowlist_sha256 {
anyhow::bail!(
"profile rank-decisions eligible-author digest {} does not match crawl-policy \
allowlist digest {}",
trusted.evidence.eligible_authors_sha256,
policy_author_allowlist_sha256
);
}
if trusted.evidence.eligible_count != policy_author_count {
anyhow::bail!(
"profile rank-decisions eligible count {} does not match crawl-policy author count {}",
trusted.evidence.eligible_count,
policy_author_count
);
}
Ok(())
}
fn require_subject_profile_rank_authority(
expected: Option<&ProfileDistanceProvenance>,
trusted: Option<&TrustedProfileRankDecisions>,
) -> Result<()> {
let Some(expected) = expected else {
return Ok(());
};
let trusted = trusted
.context("typed v3 audit subject requires its exact sealed profile-rank file and report")?;
if &trusted.evidence != expected {
anyhow::bail!(
"profile-rank evidence differs from the exact authority sealed by the audit subject"
);
}
Ok(())
}
fn manifest_root_for_index(
manifest: &hashtree_nostr::NostrEventManifest,
index: NostrEventIndex,
) -> Option<&Cid> {
match index {
NostrEventIndex::ById => manifest.by_id.as_ref(),
NostrEventIndex::ByAuthorTime => manifest.by_author_time.as_ref(),
NostrEventIndex::ByAuthorKindTime => manifest.by_author_kind_time.as_ref(),
NostrEventIndex::ByKindTime => manifest.by_kind_time.as_ref(),
NostrEventIndex::ByKindTimeAuthor => manifest.by_kind_time_author.as_ref(),
NostrEventIndex::ByTime => manifest.by_time.as_ref(),
NostrEventIndex::ByTag => manifest.by_tag.as_ref(),
NostrEventIndex::Replaceable => manifest.replaceable.as_ref(),
NostrEventIndex::ParameterizedReplaceable => manifest.parameterized_replaceable.as_ref(),
}
}
struct IndexAuditAccumulators<'a> {
expected_entries: &'a mut [EntrySetProof; 9],
missing_event_blobs: Option<&'a mut Vec<BulkProjectionMissingEventBlob>>,
max_missing_event_blobs: usize,
}
async fn audit_index_root(
spool: &BulkProjectionSpool,
target: &NostrEventStore<ReadOnlyPoolStore>,
btree: &BTree<ReadOnlyPoolStore>,
index: NostrEventIndex,
root: Option<&Cid>,
page_size: usize,
accumulators: &mut IndexAuditAccumulators<'_>,
) -> Result<(BulkProjectionIndexAudit, EntrySetProof)> {
let mut digest = Sha256::new();
digest.update(b"hashtree-nostr-bulk-index-parity-v1\0");
digest.update(index.name().as_bytes());
digest.update([0]);
let mut retained_set = EntrySetProof::default();
let mut cursor = EntryTrieCursor::new(spool, index);
let Some(root) = root else {
if let Some((key, _)) = cursor.next_entry()? {
anyhow::bail!(
"manifest {} root is empty but spool starts with `{key}`",
index.name()
);
}
return Ok((
BulkProjectionIndexAudit {
index: index.name().to_string(),
root: None,
nodes: 0,
links: 0,
durable_values_validated: 0,
entries_sha256: hex::encode(digest.finalize()),
retained_set_sha256: retained_set.evidence_sha256(),
first_key: None,
last_key: None,
},
retained_set,
));
};
let structural = btree
.validate_link_tree(Some(root))
.await
.with_context(|| format!("exhaustively validate {} link tree", index.name()))?;
let mut after = None::<String>;
let mut links = 0u64;
let mut durable_values_validated = 0u64;
let mut first_key = None;
let mut last_key = None;
loop {
let page = btree
.range_links_limited_after(root, after.as_deref(), None, page_size)
.await
.with_context(|| format!("read {} root parity page", index.name()))?;
if page.is_empty() {
break;
}
for (key, cid) in &page {
let Some((spool_key, spool_cid)) = cursor.next_entry()? else {
anyhow::bail!(
"{} root has an extra key `{key}` at row {links}",
index.name()
);
};
if spool_key != *key {
anyhow::bail!(
"{} key mismatch at row {links}: spool=`{spool_key}` root=`{key}`",
index.name()
);
}
if spool_cid != *cid {
anyhow::bail!("{} CID mismatch at row {links}, key=`{key}`", index.name());
}
retained_set.insert(index, key, cid)?;
if index == NostrEventIndex::ById {
let record = spool
.event_record(key)?
.with_context(|| format!("by-id spool key `{key}` has no event record"))?;
let record_cid = Cid {
hash: record.cid_hash,
key: record.cid_key,
};
if record_cid != *cid {
anyhow::bail!("by-id event record CID differs at key `{key}`");
}
match accumulators.missing_event_blobs.as_deref_mut() {
Some(missing) => {
match target.try_load_event_blob(cid).await.with_context(|| {
format!("exhaustively probe durable by-id event `{key}`")
})? {
Some(durable) => {
if durable.id != *key || durable != record.event {
anyhow::bail!(
"durable by-id event `{key}` differs from its exact spool record"
);
}
durable_values_validated =
durable_values_validated.checked_add(1).context(
"bulk index durable value validation count overflow",
)?;
}
None => {
if missing.len() >= accumulators.max_missing_event_blobs {
anyhow::bail!(
"event-blob repair missing set exceeds explicit limit {}",
accumulators.max_missing_event_blobs
);
}
missing.push(BulkProjectionMissingEventBlob {
event_id: key.clone(),
cid: cid.clone(),
});
}
}
}
None => {
let durable = target.load_event_blob(cid).await.with_context(|| {
format!("exhaustively load durable by-id event `{key}`")
})?;
if durable.id != *key || durable != record.event {
anyhow::bail!(
"durable by-id event `{key}` differs from its exact spool record"
);
}
durable_values_validated = durable_values_validated
.checked_add(1)
.context("bulk index durable value validation count overflow")?;
}
}
let entries = nostr_event_index_entries(&record.event, cid);
for (position, entry) in entries.iter().enumerate() {
if entries[..position]
.iter()
.any(|seen| seen.index == entry.index && seen.key == entry.key)
{
continue;
}
accumulators.expected_entries[entry.index.stable_id() as usize].insert(
entry.index,
&entry.key,
&entry.cid,
)?;
}
}
digest.update((key.len() as u64).to_be_bytes());
digest.update(key.as_bytes());
let encoded_cid = encode_cid(cid);
digest.update((encoded_cid.len() as u64).to_be_bytes());
digest.update(encoded_cid);
if first_key.is_none() {
first_key = Some(key.clone());
}
last_key = Some(key.clone());
links = links
.checked_add(1)
.context("bulk index parity link count overflow")?;
}
after = Some(page.last().expect("non-empty index parity page").0.clone());
if page.len() < page_size {
break;
}
}
if let Some((key, _)) = cursor.next_entry()? {
anyhow::bail!(
"{} root ended at row {links} before spool key `{key}`",
index.name()
);
}
if links != structural.links {
anyhow::bail!(
"{} structural link count {} differs from exact parity count {links}",
index.name(),
structural.links
);
}
Ok((
BulkProjectionIndexAudit {
index: index.name().to_string(),
root: Some(cid_to_nhash(root)?),
nodes: structural.nodes,
links,
durable_values_validated,
entries_sha256: hex::encode(digest.finalize()),
retained_set_sha256: retained_set.evidence_sha256(),
first_key,
last_key,
},
retained_set,
))
}
fn exact_index_parity_evidence_sha256(
btree_order: usize,
spool_event_records: u64,
indexes: &[BulkProjectionExactIndexParityRoot],
) -> Result<String> {
let encoded = serde_json::to_vec(&(
EXACT_INDEX_PARITY_FORMAT,
btree_order,
spool_event_records,
indexes,
))
.context("encode exact event-index parity evidence")?;
Ok(bytes_sha256(&encoded))
}
pub(super) fn validate_exact_event_index_parity_evidence(
evidence: &BulkProjectionExactIndexParityEvidence,
built_roots: &BTreeMap<u8, String>,
btree_order: usize,
) -> Result<()> {
validate_complete_built_roots("exact parity authority", built_roots)?;
if evidence.format != EXACT_INDEX_PARITY_FORMAT
|| evidence.btree_order != btree_order
|| evidence.indexes.len() != NostrEventIndex::ALL.len()
{
anyhow::bail!("exact event-index parity evidence has invalid format or dimensions");
}
for (position, index) in NostrEventIndex::ALL.iter().copied().enumerate() {
let root = built_roots
.get(&index.stable_id())
.expect("all exact parity roots checked");
let audited = &evidence.indexes[position];
if audited.index != index.name()
|| audited.stable_id != index.stable_id()
|| &audited.root != root
{
anyhow::bail!(
"exact event-index parity evidence differs for {}",
index.name()
);
}
require_sha256_text(
&format!("{} ordered-entry evidence", index.name()),
&audited.ordered_entries_sha256,
)?;
require_sha256_text(
&format!("{} retained-set evidence", index.name()),
&audited.retained_set_sha256,
)?;
}
if evidence.spool_event_records
!= evidence
.indexes
.first()
.context("exact parity evidence omitted by-id index")?
.durable_values_validated
{
anyhow::bail!("exact parity spool event count differs from durable by-id validation count");
}
let expected_sha256 = exact_index_parity_evidence_sha256(
evidence.btree_order,
evidence.spool_event_records,
&evidence.indexes,
)?;
if evidence.evidence_sha256 != expected_sha256 {
anyhow::bail!("exact event-index parity evidence SHA-256 is invalid");
}
Ok(())
}
async fn audit_exact_event_index_parity_with_details(
spool: &BulkProjectionSpool,
store: Arc<ReadOnlyPoolStore>,
built_roots: &BTreeMap<u8, String>,
btree_order: usize,
page_size: usize,
) -> Result<(
BulkProjectionExactIndexParityEvidence,
Vec<BulkProjectionIndexAudit>,
)> {
if btree_order < 2 {
anyhow::bail!("exact event-index parity requires B-tree order at least 2");
}
if page_size == 0 {
anyhow::bail!("exact event-index parity page size must be non-zero");
}
validate_complete_built_roots("exact parity authority", built_roots)?;
if NostrEventIndex::ALL.first().copied() != Some(NostrEventIndex::ById) {
anyhow::bail!("exact event-index parity requires by-id to be the first canonical index");
}
let target = NostrEventStore::new(Arc::clone(&store));
let btree = BTree::new(
store,
BTreeOptions {
order: Some(btree_order),
},
);
let mut expected_entries = [EntrySetProof::default(); 9];
let mut details = Vec::with_capacity(NostrEventIndex::ALL.len());
let mut indexes = Vec::with_capacity(NostrEventIndex::ALL.len());
for index in NostrEventIndex::ALL {
let encoded = built_roots
.get(&index.stable_id())
.expect("all exact parity roots checked");
let root = parse_root_text(encoded)
.with_context(|| format!("parse exact parity {} root", index.name()))?;
let (audit, retained_set) = {
let mut accumulators = IndexAuditAccumulators {
expected_entries: &mut expected_entries,
missing_event_blobs: None,
max_missing_event_blobs: 0,
};
audit_index_root(
spool,
&target,
&btree,
index,
Some(&root),
page_size,
&mut accumulators,
)
.await?
};
let expected = expected_entries[index.stable_id() as usize];
if retained_set != expected {
anyhow::bail!(
"{} root/spool entries do not exactly match the retained by-id event set: \
actual_count={} expected_count={} actual_digest={} expected_digest={}",
index.name(),
retained_set.count,
expected.count,
retained_set.evidence_sha256(),
expected.evidence_sha256()
);
}
indexes.push(BulkProjectionExactIndexParityRoot {
index: index.name().to_string(),
stable_id: index.stable_id(),
root: encoded.clone(),
nodes: audit.nodes,
entries: audit.links,
durable_values_validated: audit.durable_values_validated,
ordered_entries_sha256: audit.entries_sha256.clone(),
retained_set_sha256: audit.retained_set_sha256.clone(),
});
details.push(audit);
}
let spool_event_records = spool.event_record_count()?;
if spool_event_records != indexes[0].durable_values_validated {
anyhow::bail!(
"bulk spool contains {spool_event_records} event records but by-id validated {}",
indexes[0].durable_values_validated
);
}
let evidence_sha256 =
exact_index_parity_evidence_sha256(btree_order, spool_event_records, &indexes)?;
let evidence = BulkProjectionExactIndexParityEvidence {
format: EXACT_INDEX_PARITY_FORMAT.to_string(),
btree_order,
spool_event_records,
indexes,
evidence_sha256,
};
validate_exact_event_index_parity_evidence(&evidence, built_roots, btree_order)?;
Ok((evidence, details))
}
pub(super) async fn audit_exact_event_index_parity(
spool: &BulkProjectionSpool,
store: Arc<ReadOnlyPoolStore>,
built_roots: &BTreeMap<u8, String>,
btree_order: usize,
page_size: usize,
) -> Result<BulkProjectionExactIndexParityEvidence> {
Ok(audit_exact_event_index_parity_with_details(
spool,
store,
built_roots,
btree_order,
page_size,
)
.await?
.0)
}
pub(super) async fn audit_event_index_layout_and_collect_missing_blobs(
spool: &BulkProjectionSpool,
store: Arc<ReadOnlyPoolStore>,
built_roots: &BTreeMap<u8, String>,
btree_order: usize,
page_size: usize,
max_missing_event_blobs: usize,
) -> Result<(u64, Vec<BulkProjectionMissingEventBlob>)> {
if btree_order < 2 {
anyhow::bail!("event-blob repair layout audit requires B-tree order at least 2");
}
if page_size == 0 {
anyhow::bail!("event-blob repair layout audit page size must be non-zero");
}
if max_missing_event_blobs == 0 {
anyhow::bail!("event-blob repair missing-set limit must be non-zero");
}
validate_complete_built_roots("event-blob repair layout authority", built_roots)?;
if NostrEventIndex::ALL.first().copied() != Some(NostrEventIndex::ById) {
anyhow::bail!(
"event-blob repair layout audit requires by-id to be the first canonical index"
);
}
let target = NostrEventStore::new(Arc::clone(&store));
let btree = BTree::new(
store,
BTreeOptions {
order: Some(btree_order),
},
);
let mut expected_entries = [EntrySetProof::default(); 9];
let mut missing = Vec::new();
let mut durable_by_id = None::<u64>;
for index in NostrEventIndex::ALL {
let encoded = built_roots
.get(&index.stable_id())
.expect("all event-blob repair layout roots checked");
let root = parse_root_text(encoded)
.with_context(|| format!("parse event-blob repair {} root", index.name()))?;
let (audit, retained_set) = {
let mut accumulators = IndexAuditAccumulators {
expected_entries: &mut expected_entries,
missing_event_blobs: Some(&mut missing),
max_missing_event_blobs,
};
audit_index_root(
spool,
&target,
&btree,
index,
Some(&root),
page_size,
&mut accumulators,
)
.await?
};
let expected = expected_entries[index.stable_id() as usize];
if retained_set != expected {
anyhow::bail!(
"{} root/spool entries do not exactly match the retained by-id event set: \
actual_count={} expected_count={} actual_digest={} expected_digest={}",
index.name(),
retained_set.count,
expected.count,
retained_set.evidence_sha256(),
expected.evidence_sha256()
);
}
if index == NostrEventIndex::ById {
durable_by_id = Some(audit.durable_values_validated);
}
}
let spool_event_records = spool.event_record_count()?;
let accounted = durable_by_id
.context("event-blob repair layout audit omitted by-id validation count")?
.checked_add(missing.len() as u64)
.context("event-blob repair layout accounting overflow")?;
if accounted != spool_event_records {
anyhow::bail!(
"bulk spool contains {spool_event_records} event records but the layout audit accounted for {accounted}"
);
}
for pair in missing.windows(2) {
if pair[0].event_id >= pair[1].event_id {
anyhow::bail!("event-blob repair layout produced a non-unique missing event set");
}
}
Ok((spool_event_records, missing))
}
async fn load_spool_prefix_events(
spool: &BulkProjectionSpool,
target: &NostrEventStore<ReadOnlyPoolStore>,
index: NostrEventIndex,
prefix: &str,
limit: usize,
) -> Result<Vec<StoredNostrEvent>> {
let mut cursor = EntryTrieCursor::new(spool, index);
let mut events = Vec::new();
while let Some((key, cid)) = cursor.next_entry()? {
if !key.starts_with(prefix) {
if !events.is_empty() {
break;
}
continue;
}
events.push(
target
.load_event_blob(&cid)
.await
.with_context(|| format!("load {} query parity event `{key}`", index.name()))?,
);
if events.len() == limit {
break;
}
}
Ok(events)
}
fn event_ids(events: &[StoredNostrEvent]) -> Vec<String> {
events.iter().map(|event| event.id.clone()).collect()
}
fn checked_query(
query: &str,
parameters: serde_json::Value,
expected: &[StoredNostrEvent],
actual: &[StoredNostrEvent],
) -> Result<BulkProjectionQueryAudit> {
let expected_ids = event_ids(expected);
let actual_ids = event_ids(actual);
if actual_ids != expected_ids {
anyhow::bail!(
"{query} query differs from deterministic spool truth: expected={expected_ids:?} actual={actual_ids:?}"
);
}
Ok(BulkProjectionQueryAudit {
query: query.to_string(),
parameters,
event_ids: actual_ids,
})
}
fn first_spool_entry(spool: &BulkProjectionSpool, index: NostrEventIndex) -> Result<(String, Cid)> {
EntryTrieCursor::new(spool, index)
.next_entry()?
.with_context(|| format!("{} spool has no real query candidate", index.name()))
}
async fn audit_real_queries(
spool: &BulkProjectionSpool,
target: &NostrEventStore<ReadOnlyPoolStore>,
root: &Cid,
limit: usize,
) -> Result<(Vec<BulkProjectionQueryAudit>, Cid)> {
let list_options = || ListEventsOptions {
limit: Some(limit),
since: None,
until: None,
};
let mut queries = Vec::new();
let (by_id_key, representative_event_cid) = first_spool_entry(spool, NostrEventIndex::ById)?;
let expected_by_id =
load_spool_prefix_events(spool, target, NostrEventIndex::ById, &by_id_key, 1).await?;
let actual_by_id = target
.get_by_id(Some(root), &by_id_key)
.await
.context("query by-id terminal candidate")?
.into_iter()
.collect::<Vec<_>>();
queries.push(checked_query(
"by-id",
serde_json::json!({"id": by_id_key}),
&expected_by_id,
&actual_by_id,
)?);
let (_, author_cid) = first_spool_entry(spool, NostrEventIndex::ByAuthorTime)?;
let author_event = target.load_event_blob(&author_cid).await?;
let author_prefix = format!("{}:", author_event.pubkey);
let expected_author = load_spool_prefix_events(
spool,
target,
NostrEventIndex::ByAuthorTime,
&author_prefix,
limit,
)
.await?;
let actual_author = target
.list_by_author(Some(root), &author_event.pubkey, list_options())
.await?;
queries.push(checked_query(
"by-author",
serde_json::json!({"author": author_event.pubkey, "limit": limit}),
&expected_author,
&actual_author,
)?);
let (_, author_kind_cid) = first_spool_entry(spool, NostrEventIndex::ByAuthorKindTime)?;
let author_kind_event = target.load_event_blob(&author_kind_cid).await?;
let author_kind_prefix = format!(
"{}:{:08x}:",
author_kind_event.pubkey, author_kind_event.kind
);
let expected_author_kind = load_spool_prefix_events(
spool,
target,
NostrEventIndex::ByAuthorKindTime,
&author_kind_prefix,
limit,
)
.await?;
let actual_author_kind = target
.list_by_author_and_kind(
Some(root),
&author_kind_event.pubkey,
author_kind_event.kind,
list_options(),
)
.await?;
queries.push(checked_query(
"by-author-kind",
serde_json::json!({
"author": author_kind_event.pubkey,
"kind": author_kind_event.kind,
"limit": limit
}),
&expected_author_kind,
&actual_author_kind,
)?);
let (_, kind_cid) = first_spool_entry(spool, NostrEventIndex::ByKindTime)?;
let kind_event = target.load_event_blob(&kind_cid).await?;
let kind_prefix = format!("{:08x}:", kind_event.kind);
let expected_kind = load_spool_prefix_events(
spool,
target,
NostrEventIndex::ByKindTime,
&kind_prefix,
limit,
)
.await?;
let actual_kind = target
.list_by_kind(Some(root), kind_event.kind, list_options())
.await?;
queries.push(checked_query(
"by-kind",
serde_json::json!({"kind": kind_event.kind, "limit": limit}),
&expected_kind,
&actual_kind,
)?);
let expected_recent =
load_spool_prefix_events(spool, target, NostrEventIndex::ByTime, "", limit).await?;
let actual_recent = target.list_recent(Some(root), list_options()).await?;
queries.push(checked_query(
"recent",
serde_json::json!({"limit": limit}),
&expected_recent,
&actual_recent,
)?);
let (tag_key, _) = first_spool_entry(spool, NostrEventIndex::ByTag)?;
let tag_prefix = tag_key
.rsplit_once(':')
.and_then(|(without_id, _)| without_id.rsplit_once(':').map(|(prefix, _)| prefix))
.context("first by-tag spool key has no timestamp/event suffix")?;
let (tag_name, tag_value) = tag_prefix
.split_once(':')
.context("first by-tag spool key has no name/value prefix")?;
let expected_tag = load_spool_prefix_events(
spool,
target,
NostrEventIndex::ByTag,
&format!("{tag_prefix}:"),
limit,
)
.await?;
let actual_tag = target
.list_by_tag(Some(root), tag_name, tag_value, list_options())
.await?;
queries.push(checked_query(
"by-tag",
serde_json::json!({
"tag": tag_name,
"value": tag_value,
"limit": limit
}),
&expected_tag,
&actual_tag,
)?);
let (_, replaceable_cid) = first_spool_entry(spool, NostrEventIndex::Replaceable)?;
let replaceable_event = target.load_event_blob(&replaceable_cid).await?;
let expected_replaceable = vec![replaceable_event.clone()];
let actual_replaceable = target
.get_replaceable(
Some(root),
&replaceable_event.pubkey,
replaceable_event.kind,
)
.await?
.into_iter()
.collect::<Vec<_>>();
queries.push(checked_query(
"replaceable",
serde_json::json!({
"author": replaceable_event.pubkey,
"kind": replaceable_event.kind
}),
&expected_replaceable,
&actual_replaceable,
)?);
let (_, parameterized_cid) =
first_spool_entry(spool, NostrEventIndex::ParameterizedReplaceable)?;
let parameterized_event = target.load_event_blob(¶meterized_cid).await?;
let d_tag = parameterized_event
.tags
.iter()
.find_map(|tag| match tag.as_slice() {
[name, value, ..] if name == "d" => Some(value.as_str()),
_ => None,
})
.unwrap_or_default();
let expected_parameterized = vec![parameterized_event.clone()];
let actual_parameterized = target
.get_parameterized_replaceable(
Some(root),
¶meterized_event.pubkey,
parameterized_event.kind,
d_tag,
)
.await?
.into_iter()
.collect::<Vec<_>>();
queries.push(checked_query(
"parameterized-replaceable",
serde_json::json!({
"author": parameterized_event.pubkey,
"kind": parameterized_event.kind,
"d": d_tag
}),
&expected_parameterized,
&actual_parameterized,
)?);
Ok((queries, representative_event_cid))
}
async fn block_evidence(
role: impl Into<String>,
cid: &Cid,
store: &ReadOnlyPoolStore,
) -> Result<BulkProjectionBlockEvidence> {
store
.get(&cid.hash)
.await
.with_context(|| format!("read representative block {}", hex::encode(cid.hash)))?
.with_context(|| format!("representative block {} is missing", hex::encode(cid.hash)))?;
Ok(BulkProjectionBlockEvidence {
role: role.into(),
nhash: cid_to_nhash(cid)?,
sha256: hex::encode(cid.hash),
})
}
fn validate_profile_search_value(
key: &str,
value: &str,
expected: &ExpectedProfileSearchEntry,
bound_values: &mut BTreeMap<String, hashtree_cli::socialgraph::StoredProfileSearchEntry>,
) -> Result<hashtree_cli::socialgraph::StoredProfileSearchEntry> {
let actual: StrictStoredProfileSearchEntry = serde_json::from_str(value)
.with_context(|| format!("strictly decode profile-search entry `{key}`"))?;
let actual: hashtree_cli::socialgraph::StoredProfileSearchEntry = actual.into();
if actual.pubkey != expected.pubkey {
anyhow::bail!(
"profile-search entry `{key}` names pubkey `{}` instead of `{}`",
actual.pubkey,
expected.pubkey
);
}
let bound_distance = bound_values
.get(&expected.pubkey)
.map(|entry| entry.follow_distance)
.unwrap_or(actual.follow_distance);
let reconstructed = hashtree_cli::socialgraph::stored_profile_search_entry_for_event(
&expected.event,
&expected.mirrored_cid,
bound_distance,
)
.with_context(|| format!("reconstruct exact profile-search entry `{key}`"))?;
if actual != reconstructed {
anyhow::bail!("profile-search entry `{key}` differs from its exact builder reconstruction");
}
if let Some(bound) = bound_values.get(&expected.pubkey) {
if actual != *bound {
anyhow::bail!(
"profile-search entry `{key}` is not identical across all keys for pubkey `{}`",
expected.pubkey
);
}
} else {
bound_values.insert(expected.pubkey.clone(), actual.clone());
}
Ok(actual)
}
fn require_profile_distance_attestation(
allow_recovery_tranche: bool,
expected: Option<&str>,
actual: &str,
has_independent_provenance: bool,
) -> Result<()> {
if let Some(expected) = expected {
require_expected_sha256("profile follow-distance seal", actual, expected)?;
}
if !allow_recovery_tranche && !has_independent_provenance {
anyhow::bail!(
"full-policy cutover audit requires independently pinned profile rank-decisions \
provenance; an opaque v2 distance seal cannot self-attest"
);
}
Ok(())
}
fn validate_bound_profile_distances(
bound_values: &BTreeMap<String, hashtree_cli::socialgraph::StoredProfileSearchEntry>,
trusted_decisions: &BTreeMap<String, Option<u32>>,
) -> Result<()> {
for (pubkey, entry) in bound_values {
match trusted_decisions.get(pubkey) {
Some(Some(distance)) if entry.follow_distance == Some(*distance) => {}
Some(Some(distance)) => anyhow::bail!(
"profile-search retained pubkey `{pubkey}` with distance {:?}, \
but trusted rank decision requires {distance}",
entry.follow_distance
),
Some(None) => anyhow::bail!(
"profile-search retained pubkey `{pubkey}` despite its trusted exclusion decision"
),
None => anyhow::bail!(
"profile-search retained pubkey `{pubkey}` without a trusted rank decision"
),
}
}
Ok(())
}
pub(super) async fn audit_profile_indexes_at_roots<S: Store>(
spool: &BulkProjectionSpool,
store: Arc<S>,
roots_before: &hashtree_cli::socialgraph::ProfileIndexRoots,
trusted_rank_decisions: Option<&BTreeMap<String, Option<u32>>>,
) -> Result<(BulkProjectionProfileAudit, Vec<Cid>)> {
let by_pubkey_root = roots_before
.by_pubkey
.clone()
.context("profile-by-pubkey root is missing")?;
let by_pubkey_root_file_sha256 = roots_before
.by_pubkey_file_sha256
.clone()
.context("profile-by-pubkey root file hash is missing")?;
let search_root = roots_before
.search
.clone()
.context("profile-search root is missing")?;
let search_root_file_sha256 = roots_before
.search_file_sha256
.clone()
.context("profile-search root file hash is missing")?;
let btree = BTree::new(Arc::clone(&store), BTreeOptions { order: Some(64) });
let by_pubkey = btree
.validate_link_tree(Some(&by_pubkey_root))
.await
.context("exhaustively validate profile-by-pubkey root")?;
let by_pubkey_entries = btree
.links_entries(Some(&by_pubkey_root))
.await
.context("exhaustively traverse profile-by-pubkey root")?;
if by_pubkey_entries.len() as u64 != by_pubkey.links {
anyhow::bail!(
"profile-by-pubkey structural count {} differs from traversal count {}",
by_pubkey.links,
by_pubkey_entries.len()
);
}
let mut retained_profiles = spool.retained_profile_records()?;
if retained_profiles.is_empty() {
anyhow::bail!("bulk spool contains no retained kind-0 profiles");
}
let tree = HashTree::new(HashTreeConfig::new(Arc::clone(&store)));
let mut by_pubkey_digest = Sha256::new();
by_pubkey_digest.update(b"hashtree-profile-by-pubkey-audit-v1\0");
let mut expected_search = BTreeMap::<String, ExpectedProfileSearchEntry>::new();
let mut representative_profile_cid = None;
for (pubkey, mirrored_cid) in &by_pubkey_entries {
let record = retained_profiles
.remove(pubkey)
.with_context(|| format!("profile-by-pubkey contains unexpected pubkey `{pubkey}`"))?;
let expected_event = record
.event
.to_nostr_sdk_event()
.with_context(|| format!("decode retained profile event {}", record.event.id))?;
let mirrored_bytes = tree
.get(mirrored_cid, None)
.await
.with_context(|| format!("read mirrored profile event for {pubkey}"))?
.with_context(|| format!("mirrored profile event for {pubkey} is missing"))?;
let mirrored = Event::from_json(
String::from_utf8(mirrored_bytes)
.with_context(|| format!("decode mirrored profile event for {pubkey} as UTF-8"))?,
)
.with_context(|| format!("decode mirrored profile event JSON for {pubkey}"))?;
if mirrored != expected_event {
anyhow::bail!(
"profile-by-pubkey mirrored event for `{pubkey}` differs from retained event {}",
record.event.id
);
}
for key in hashtree_cli::socialgraph::profile_search_keys_for_event(&mirrored) {
let expected = ExpectedProfileSearchEntry {
pubkey: pubkey.clone(),
event: mirrored.clone(),
mirrored_cid: mirrored_cid.clone(),
};
if expected_search.insert(key.clone(), expected).is_some() {
anyhow::bail!("retained profiles produced duplicate search key `{key}`");
}
}
by_pubkey_digest.update((pubkey.len() as u64).to_be_bytes());
by_pubkey_digest.update(pubkey.as_bytes());
let encoded_cid = encode_cid(mirrored_cid);
by_pubkey_digest.update((encoded_cid.len() as u64).to_be_bytes());
by_pubkey_digest.update(encoded_cid);
representative_profile_cid.get_or_insert_with(|| mirrored_cid.clone());
}
if !retained_profiles.is_empty() {
anyhow::bail!(
"profile-by-pubkey omitted {} retained profiles; first missing pubkey `{}`",
retained_profiles.len(),
retained_profiles
.first_key_value()
.expect("non-empty retained profiles")
.0
);
}
let search_structural = btree
.validate_value_tree(Some(&search_root))
.await
.context("exhaustively validate profile-search root structure")?;
let search_entries = btree
.entries(Some(&search_root))
.await
.context("exhaustively traverse profile-search root")?;
if search_entries.len() as u64 != search_structural.entries {
anyhow::bail!(
"profile-search structural count {} differs from traversal count {}",
search_structural.entries,
search_entries.len()
);
}
let mut search_digest = Sha256::new();
search_digest.update(b"hashtree-profile-search-audit-v1\0");
let mut bound_values = BTreeMap::new();
let mut sample = None;
for (key, value) in &search_entries {
let expected = expected_search
.remove(key)
.with_context(|| format!("profile-search contains unexpected key `{key}`"))?;
let entry = validate_profile_search_value(key, value, &expected, &mut bound_values)?;
sample.get_or_insert_with(|| {
(
expected.pubkey.clone(),
expected.event.id.to_hex(),
entry.name.clone(),
)
});
search_digest.update((key.len() as u64).to_be_bytes());
search_digest.update(key.as_bytes());
search_digest.update((value.len() as u64).to_be_bytes());
search_digest.update(value.as_bytes());
}
if !expected_search.is_empty() {
anyhow::bail!(
"profile-search omitted {} expected keys; first missing key `{}`",
expected_search.len(),
expected_search
.first_key_value()
.expect("non-empty expected search entries")
.0
);
}
if bound_values.len() != by_pubkey_entries.len() {
anyhow::bail!(
"profile-search bound {} historic distances for {} retained profiles",
bound_values.len(),
by_pubkey_entries.len()
);
}
if let Some(trusted_rank_decisions) = trusted_rank_decisions {
validate_bound_profile_distances(&bound_values, trusted_rank_decisions)?;
}
let bound_distances = bound_values
.iter()
.map(|(pubkey, entry)| (pubkey.clone(), entry.follow_distance))
.collect();
let follow_distance_seal_sha256 =
hashtree_cli::socialgraph::profile_follow_distance_seal_v2(&bound_distances);
let (sample_pubkey, sample_event_id, sample_name) =
sample.context("profile-search root contains no entries")?;
let representative_profile_cid =
representative_profile_cid.context("profile-by-pubkey root contains no entries")?;
let follow_distance_binding = if trusted_rank_decisions.is_some() {
"independently-pinned-complete-rank-decisions@1"
} else {
"opaque-historic-v2-first-observed-per-pubkey; \
v3-requires-deterministic-or-attested-distance"
};
Ok((
BulkProjectionProfileAudit {
by_pubkey_root: cid_to_nhash(&by_pubkey_root)?,
by_pubkey_root_file_sha256,
by_pubkey_nodes: by_pubkey.nodes,
by_pubkey_links: by_pubkey.links,
by_pubkey_entries_sha256: hex::encode(by_pubkey_digest.finalize()),
search_root: cid_to_nhash(&search_root)?,
search_root_file_sha256,
search_nodes: search_structural.nodes,
search_entries: search_entries.len() as u64,
search_entries_sha256: hex::encode(search_digest.finalize()),
sample_pubkey,
sample_event_id,
sample_name,
follow_distance_binding: follow_distance_binding.to_string(),
follow_distance_seal_sha256,
},
vec![by_pubkey_root, search_root, representative_profile_cid],
))
}
async fn audit_profile_indexes(
data_dir: &Path,
spool: &BulkProjectionSpool,
store: Arc<ReadOnlyPoolStore>,
trusted_rank_decisions: Option<&BTreeMap<String, Option<u32>>>,
) -> Result<(
BulkProjectionProfileAudit,
Vec<Cid>,
hashtree_cli::socialgraph::ProfileIndexRoots,
)> {
let roots_before = hashtree_cli::socialgraph::read_profile_index_roots(data_dir)?;
let (audit, cids) =
audit_profile_indexes_at_roots(spool, store, &roots_before, trusted_rank_decisions).await?;
let roots_after = hashtree_cli::socialgraph::read_profile_index_roots(data_dir)?;
if roots_after != roots_before {
anyhow::bail!("profile index root files changed during read-only audit");
}
Ok((audit, cids, roots_before))
}
#[derive(Debug)]
struct PreparedAuditOutput {
path: PathBuf,
parent: PathBuf,
}
fn prepare_audit_output_path(
data_dir: &Path,
staging_data_dir: &Path,
out: &Path,
) -> Result<PreparedAuditOutput> {
if !out.is_absolute() {
anyhow::bail!("bulk projection audit --out must be an absolute path");
}
let file_name = out
.file_name()
.filter(|name| !name.is_empty())
.context("bulk projection audit --out must name a file")?;
let parent = out
.parent()
.context("bulk projection audit --out has no parent directory")?
.canonicalize()
.with_context(|| {
format!(
"canonicalize pre-existing audit output parent {}",
out.parent().expect("parent checked").display()
)
})?;
let data_tree = data_dir
.canonicalize()
.with_context(|| format!("canonicalize audited data tree {}", data_dir.display()))?;
let staging_tree = staging_data_dir.canonicalize().with_context(|| {
format!(
"canonicalize audited staging tree {}",
staging_data_dir.display()
)
})?;
if parent.starts_with(&data_tree) || parent.starts_with(&staging_tree) {
anyhow::bail!(
"bulk projection audit evidence parent {} is inside an audited live data tree",
parent.display()
);
}
let path = parent.join(file_name);
match std::fs::symlink_metadata(&path) {
Ok(_) => anyhow::bail!(
"bulk projection audit evidence target already exists or is a symlink: {}",
path.display()
),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => {
return Err(error)
.with_context(|| format!("inspect audit evidence target {}", path.display()));
}
}
Ok(PreparedAuditOutput { path, parent })
}
fn sync_directory(path: &Path) -> Result<()> {
File::open(path)
.with_context(|| format!("open directory {} for evidence fsync", path.display()))?
.sync_all()
.with_context(|| format!("fsync evidence directory {}", path.display()))
}
fn install_audit_evidence_noreplace(bytes: &[u8], prepared: &PreparedAuditOutput) -> Result<()> {
let mut temp_path = None;
let mut temp_file = None;
for _ in 0..16 {
let candidate = prepared.parent.join(format!(
".bulk-projection-audit.{}.{}.tmp",
std::process::id(),
uuid::Uuid::new_v4()
));
match OpenOptions::new()
.write(true)
.create_new(true)
.open(&candidate)
{
Ok(file) => {
temp_path = Some(candidate);
temp_file = Some(file);
break;
}
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => continue,
Err(error) => {
return Err(error).with_context(|| {
format!(
"create same-directory audit evidence temp file {}",
candidate.display()
)
});
}
}
}
let temp_path = temp_path.context("could not allocate unique audit evidence temp file")?;
let mut temp_file = temp_file.expect("temp path and file are assigned together");
let write_result = (|| -> Result<()> {
temp_file
.write_all(bytes)
.context("write complete audit evidence temp file")?;
temp_file
.sync_all()
.context("fsync complete audit evidence temp file")?;
drop(temp_file);
std::fs::hard_link(&temp_path, &prepared.path).with_context(|| {
format!(
"atomically install no-replace audit evidence {}",
prepared.path.display()
)
})?;
sync_directory(&prepared.parent)?;
std::fs::remove_file(&temp_path).with_context(|| {
format!(
"remove linked audit evidence temp file {}",
temp_path.display()
)
})?;
sync_directory(&prepared.parent)?;
Ok(())
})();
if write_result.is_err() {
let _ = std::fs::remove_file(&temp_path);
let _ = sync_directory(&prepared.parent);
}
write_result
}
fn canonical_audit_output_bytes(output: &BulkProjectionAuditOutput) -> Result<Vec<u8>> {
let mut bytes =
serde_json::to_vec_pretty(output).context("serialize bulk projection audit evidence")?;
bytes.push(b'\n');
Ok(bytes)
}
fn audited_index_roots(indexes: &[BulkProjectionIndexAudit]) -> Result<BTreeMap<u8, String>> {
if indexes.len() != NostrEventIndex::ALL.len() {
anyhow::bail!("full-cutover audit must contain exactly nine indexes");
}
let mut roots = BTreeMap::new();
for audit in indexes {
let index = NostrEventIndex::ALL
.into_iter()
.find(|index| index.name() == audit.index)
.with_context(|| format!("full-cutover audit has unknown index `{}`", audit.index))?;
let root = audit
.root
.as_deref()
.with_context(|| format!("full-cutover audit has empty {} root", audit.index))?;
let parsed = parse_root_text(root)
.with_context(|| format!("parse full-cutover {} root", audit.index))?;
if cid_to_nhash(&parsed)? != root {
anyhow::bail!(
"full-cutover {} root is not canonical nhash text",
audit.index
);
}
if roots.insert(index.stable_id(), root.to_string()).is_some() {
anyhow::bail!("full-cutover audit repeats index `{}`", audit.index);
}
}
validate_complete_built_roots("full-cutover audit", &roots)?;
Ok(roots)
}
fn validate_profile_distance_provenance_for_cutover(
provenance: &ProfileDistanceProvenance,
authors_total: usize,
) -> Result<()> {
for (label, digest) in [
(
"full-cutover rank-decisions file SHA-256",
provenance.rank_decisions_file_sha256.as_str(),
),
(
"full-cutover rank-decisions report SHA-256",
provenance.rank_decisions_report_sha256.as_str(),
),
(
"full-cutover rank-decisions semantic SHA-256",
provenance.rank_decisions_sha256.as_str(),
),
(
"full-cutover social graph SHA-256",
provenance.social_graph_sha256.as_str(),
),
(
"full-cutover social graph root",
provenance.social_graph_root.as_str(),
),
(
"full-cutover eligible authors SHA-256",
provenance.eligible_authors_sha256.as_str(),
),
] {
require_sha256_text(label, digest)?;
}
if provenance.format != PROFILE_RANK_DECISION_REPORT_FORMAT
|| provenance.census_format != PROFILE_RANK_DECISION_CENSUS_FORMAT
|| provenance.rank_policy != PROFILE_RANK_POLICY
|| provenance.exclusion_policy != PROFILE_EXCLUSION_POLICY
|| provenance.overmute_threshold != PROFILE_OVERMUTE_THRESHOLD
|| provenance.eligible_count != authors_total
|| provenance
.eligible_count
.checked_add(provenance.excluded_count)
!= Some(provenance.record_count)
{
anyhow::bail!(
"full-cutover profile rank provenance does not bind the exact trusted author policy"
);
}
Ok(())
}
fn mint_full_cutover_verified(
output: &BulkProjectionAuditOutput,
evidence_bytes: &[u8],
) -> Result<Option<FullCutoverVerified>> {
if canonical_audit_output_bytes(output)? != evidence_bytes {
anyhow::bail!("full-cutover token input differs from canonical audit evidence bytes");
}
if output.recovery_tranche_only {
if output.audit_mode != RECOVERY_TRANCHE_AUDIT_MODE
|| output.cutover_eligible
|| output.authors_processed == 0
|| output.authors_processed >= output.authors_total
|| output.stage_corpus.is_some()
{
anyhow::bail!("recovery audit evidence has inconsistent non-cutover boundaries");
}
return Ok(None);
}
if output.version != BULK_PROJECTION_AUDIT_FORMAT_VERSION
|| output.audit_mode != FULL_POLICY_CUTOVER_AUDIT_MODE
|| !output.cutover_eligible
|| output.trusted_full_author_count == 0
|| output.authors_processed != output.authors_total
|| output.authors_total != output.trusted_full_author_count
{
anyhow::bail!("audit evidence does not satisfy the exact full-cutover boundary");
}
match (output.subject_kind, output.subject_version) {
(AuditSubjectKind::V3, 3) => {}
_ => anyhow::bail!("full-cutover token requires a v3 audit subject"),
}
let stage_corpus = output
.stage_corpus
.as_ref()
.context("full-cutover audit lacks sealed stage-corpus reconciliation evidence")?;
if stage_corpus.segment_count == 0
|| stage_corpus.event_cids_validated == 0
|| stage_corpus.retained_records == 0
{
anyhow::bail!("full-cutover stage-corpus evidence is empty");
}
for (label, digest) in [
(
"full-cutover retained stage records SHA-256",
stage_corpus.retained_records_sha256.as_str(),
),
(
"full-cutover stage replaceable slots SHA-256",
stage_corpus.replaceable_slots_sha256.as_str(),
),
] {
require_sha256_text(label, digest)?;
}
for (label, digest) in [
(
"full-cutover subject state SHA-256",
output.state_sha256.as_str(),
),
(
"full-cutover stage state SHA-256",
output.stage_state_sha256.as_str(),
),
(
"full-cutover policy SHA-256",
output.trusted_policy_sha256.as_str(),
),
(
"full-cutover policy author allowlist SHA-256",
output.policy_author_allowlist_sha256.as_str(),
),
(
"full-cutover PoolStore catalog SHA-256",
output.pool_catalog_sha256.as_str(),
),
(
"full-cutover PoolStore manifest SHA-256",
output.pool_manifest_sha256.as_str(),
),
(
"full-cutover profile-by-pubkey root-file SHA-256",
output.profile.by_pubkey_root_file_sha256.as_str(),
),
(
"full-cutover profile-search root-file SHA-256",
output.profile.search_root_file_sha256.as_str(),
),
(
"full-cutover profile distance seal SHA-256",
output.profile.follow_distance_seal_sha256.as_str(),
),
] {
require_sha256_text(label, digest)?;
}
if output.pool_stored_locations == 0 {
anyhow::bail!("full-cutover PoolStore evidence contains no stored locations");
}
let candidate =
parse_root_text(&output.candidate_root).context("parse full-cutover candidate root")?;
if cid_to_nhash(&candidate)? != output.candidate_root {
anyhow::bail!("full-cutover candidate root is not canonical nhash text");
}
let index_roots = audited_index_roots(&output.indexes)?;
for (label, root) in [
(
"full-cutover profile-by-pubkey root",
output.profile.by_pubkey_root.as_str(),
),
(
"full-cutover profile-search root",
output.profile.search_root.as_str(),
),
] {
let parsed = parse_root_text(root).with_context(|| format!("parse {label}"))?;
if cid_to_nhash(&parsed)? != root {
anyhow::bail!("{label} is not canonical nhash text");
}
}
let expected_distance_seal = output
.trusted_profile_distance_seal_sha256
.as_deref()
.context("full-cutover audit lacks an independently pinned profile distance seal")?;
if expected_distance_seal != output.profile.follow_distance_seal_sha256 {
anyhow::bail!("full-cutover profile distance seal differs from its trusted pin");
}
let provenance = output
.profile_distance_provenance
.as_ref()
.context("full-cutover audit lacks profile rank provenance")?;
validate_profile_distance_provenance_for_cutover(provenance, output.authors_total)?;
if provenance.eligible_authors_sha256 != output.policy_author_allowlist_sha256 {
anyhow::bail!(
"full-cutover rank provenance differs from the policy author-allowlist digest"
);
}
if output.queries.is_empty() || output.representative_blocks.is_empty() {
anyhow::bail!("full-cutover audit lacks real query or representative block evidence");
}
for block in &output.representative_blocks {
require_sha256_text(
&format!("full-cutover representative block `{}` SHA-256", block.role),
&block.sha256,
)?;
parse_root_text(&block.nhash)
.with_context(|| format!("parse full-cutover representative block `{}`", block.role))?;
}
Ok(Some(FullCutoverVerified {
evidence_sha256: bytes_sha256(evidence_bytes),
subject_kind: output.subject_kind,
subject_version: output.subject_version,
candidate_root: output.candidate_root.clone(),
subject_state_sha256: output.state_sha256.clone(),
stage_state_sha256: output.stage_state_sha256.clone(),
trusted_policy_sha256: output.trusted_policy_sha256.clone(),
policy_author_allowlist_sha256: output.policy_author_allowlist_sha256.clone(),
trusted_full_author_count: output.trusted_full_author_count,
crawl_policy_max_follow_distance: output.crawl_policy_max_follow_distance,
authors_processed: output.authors_processed,
authors_total: output.authors_total,
index_roots,
pool_catalog_sha256: output.pool_catalog_sha256.clone(),
pool_manifest_sha256: output.pool_manifest_sha256.clone(),
pool_stored_locations: output.pool_stored_locations,
stage_segment_count: stage_corpus.segment_count,
stage_event_cids_validated: stage_corpus.event_cids_validated,
stage_retained_records: stage_corpus.retained_records,
stage_retained_records_sha256: stage_corpus.retained_records_sha256.clone(),
stage_replaceable_slots: stage_corpus.replaceable_slots,
stage_replaceable_slots_sha256: stage_corpus.replaceable_slots_sha256.clone(),
profile_by_pubkey_root: output.profile.by_pubkey_root.clone(),
profile_search_root: output.profile.search_root.clone(),
profile_follow_distance_seal_sha256: output.profile.follow_distance_seal_sha256.clone(),
profile_distance_provenance: provenance.clone(),
}))
}
fn write_audit_output_noreplace(
output: &BulkProjectionAuditOutput,
prepared: &PreparedAuditOutput,
) -> Result<(Vec<u8>, Option<FullCutoverVerified>)> {
let bytes = canonical_audit_output_bytes(output)?;
let full_cutover = mint_full_cutover_verified(output, &bytes)?;
install_audit_evidence_noreplace(&bytes, prepared)?;
Ok((bytes, full_cutover))
}
pub(crate) async fn audit_bulk_projection(
data_dir: &Path,
mut options: BulkProjectionAuditOptions,
) -> Result<BulkProjectionAuditOutput> {
if options.btree_order.is_some_and(|order| order < 2)
|| options.page_size == 0
|| options.query_limit == 0
{
anyhow::bail!(
"audit B-tree order must be at least 2 and page/query limits must be non-zero"
);
}
if options.expected_full_author_count == Some(0) {
anyhow::bail!("expected full author count must be non-zero");
}
let prepared_output =
prepare_audit_output_path(data_dir, &options.staging_data_dir, &options.out)?;
let subject = if options.v3_candidate {
let caller_authority_supplied = options.expected_stage_state_sha256.is_some()
|| options.expected_policy_sha256.is_some()
|| options.expected_profile_distance_seal_sha256.is_some()
|| options.profile_rank_decisions_file.is_some()
|| options
.expected_profile_rank_decisions_file_sha256
.is_some()
|| options.profile_rank_decisions_report.is_some()
|| options
.expected_profile_rank_decisions_report_sha256
.is_some()
|| options.expected_full_author_count.is_some()
|| options.allow_recovery_tranche
|| options.btree_order.is_some();
if caller_authority_supplied {
anyhow::bail!(
"v3 Candidate audit derives stage, policy, author-count, B-tree order, \
profile-distance, and rank authority exclusively from the sealed Candidate state"
);
}
let authority = load_v3_candidate_audit_authority(
data_dir,
&options.staging_data_dir,
&options.expected_state_sha256,
)?;
if authority.author_count == 0 {
anyhow::bail!("v3 Candidate trusted full author count must be non-zero");
}
options.expected_stage_state_sha256 = Some(authority.stage_state_sha256);
options.expected_policy_sha256 = Some(authority.policy_sha256);
options.expected_profile_distance_seal_sha256 =
Some(authority.profile_distance_seal_sha256);
options.profile_rank_decisions_file = Some(authority.profile_rank_decisions_path);
options.expected_profile_rank_decisions_file_sha256 =
Some(authority.profile_rank_authority.rank_decisions_file_sha256);
options.profile_rank_decisions_report = Some(authority.profile_rank_report_path);
options.expected_profile_rank_decisions_report_sha256 = Some(
authority
.profile_rank_authority
.rank_decisions_report_sha256,
);
options.expected_full_author_count = Some(authority.author_count);
load_v3_audit_subject(
data_dir,
V3AuditSubjectSpec {
expected_state_sha256: options.expected_state_sha256.clone(),
staging_data_dir: options.staging_data_dir.clone(),
},
)?
} else {
load_v2_audit_subject(data_dir, &options)?
};
let BulkProjectionAuditCompletion {
output,
full_cutover,
} = audit_loaded_subject(data_dir, &options, subject, &prepared_output).await?;
drop(full_cutover);
Ok(output)
}
async fn audit_loaded_subject(
data_dir: &Path,
options: &BulkProjectionAuditOptions,
subject: AuditSubject,
prepared_output: &PreparedAuditOutput,
) -> Result<BulkProjectionAuditCompletion> {
let trusted_profile_rank_decisions = load_trusted_profile_rank_decisions(options)?;
require_profile_rank_policy_binding(
trusted_profile_rank_decisions.as_ref(),
&subject.policy_author_allowlist_sha256,
subject.authors_total,
)?;
require_subject_profile_rank_authority(
subject.expected_profile_distance_provenance.as_ref(),
trusted_profile_rank_decisions.as_ref(),
)?;
let spool = BulkProjectionSpool::open_read_only(&subject.spool_path)?;
let pool_path = data_dir.join(SHARED_BLOB_POOL_DIR_NAME);
let store = Arc::new(
ReadOnlyPoolStore::open(&pool_path)
.with_context(|| format!("open exact native PoolStore {}", pool_path.display()))?,
);
let catalog_before = store
.validate_committed_catalog()
.context("validate fully committed PoolStore catalog")?;
let target = NostrEventStore::new(Arc::clone(&store));
let stage_corpus = if let Some(spec) = subject.stage_corpus.as_ref() {
Some(
reconcile_v3_stage_corpus(spec, &spool, &target)
.await
.context(
"independently reconcile sealed v3 stage event-CID corpus with retained spool",
)?,
)
} else {
None
};
let canonical_manifest = target
.get_canonical_manifest(&subject.candidate_root)
.await
.context("validate exact canonical bulk projection manifest")?;
let manifest = canonical_manifest.manifest;
let manifest_metadata = canonical_manifest.metadata_cid;
let mut representative_cids = vec![subject.candidate_root.clone(), manifest_metadata];
for index in NostrEventIndex::ALL {
let encoded = subject
.built_roots
.get(&index.stable_id())
.expect("all nine state roots checked");
if encoded.is_empty() {
anyhow::bail!(
"bulk projection state omitted required canonical `{}` root",
index.name()
);
}
let state_root = Some(
parse_root_text(encoded)
.with_context(|| format!("parse state {} root", index.name()))?,
);
let manifest_root = manifest_root_for_index(&manifest, index);
if manifest_root != state_root.as_ref() {
anyhow::bail!(
"manifest {} root differs from exact projection state",
index.name()
);
}
if let Some(root) = manifest_root {
representative_cids.push(root.clone());
}
}
let (_exact_index_parity, indexes) = audit_exact_event_index_parity_with_details(
&spool,
Arc::clone(&store),
&subject.built_roots,
subject.btree_order,
options.page_size,
)
.await?;
let (profile, profile_cids, profile_roots_before) = audit_profile_indexes(
data_dir,
&spool,
Arc::clone(&store),
trusted_profile_rank_decisions
.as_ref()
.map(|trusted| &trusted.decisions),
)
.await?;
require_profile_distance_attestation(
subject.scope.recovery_tranche_only(),
subject.expected_profile_distance_seal_sha256.as_deref(),
&profile.follow_distance_seal_sha256,
trusted_profile_rank_decisions.is_some(),
)?;
representative_cids.extend(profile_cids);
let (queries, representative_event_cid) = audit_real_queries(
&spool,
&target,
&subject.candidate_root,
options.query_limit,
)
.await?;
representative_cids.push(representative_event_cid);
let mut representative_blocks = Vec::new();
let mut seen_blocks = HashSet::new();
for (position, cid) in representative_cids.iter().enumerate() {
if seen_blocks.insert(cid.hash) {
let role = match position {
0 => "manifest".to_string(),
1 => "manifest-metadata".to_string(),
2..=10 => format!("index-root-{}", position - 2),
_ => format!("representative-{position}"),
};
representative_blocks.push(block_evidence(role, cid, &store).await?);
}
}
for snapshot in &subject.snapshots {
snapshot.recheck()?;
}
let catalog_after = store
.validate_committed_catalog()
.context("revalidate PoolStore catalog after audit")?;
if catalog_after != catalog_before {
anyhow::bail!("PoolStore catalog changed during read-only audit");
}
let profile_roots_after = hashtree_cli::socialgraph::read_profile_index_roots(data_dir)?;
if profile_roots_after != profile_roots_before {
anyhow::bail!("profile index root files changed during read-only audit");
}
recheck_trusted_profile_rank_decisions(trusted_profile_rank_decisions.as_ref())?;
if let Some(spec) = subject.v3_recheck.clone() {
let reloaded = load_v3_audit_subject(data_dir, spec)
.context("revalidate typed v3 authority after exhaustive audit")?;
if reloaded != subject {
anyhow::bail!("typed v3 audit authority changed during exhaustive verification");
}
}
let output = BulkProjectionAuditOutput {
version: BULK_PROJECTION_AUDIT_FORMAT_VERSION,
subject_kind: subject.kind,
subject_version: subject.version,
candidate_root: cid_to_nhash(&subject.candidate_root)?,
state_sha256: subject.state_sha256,
stage_state_sha256: subject.stage_state_sha256,
trusted_policy_sha256: subject.trusted_policy_sha256,
policy_author_allowlist_sha256: subject.policy_author_allowlist_sha256,
trusted_profile_distance_seal_sha256: subject.expected_profile_distance_seal_sha256.clone(),
profile_distance_provenance: trusted_profile_rank_decisions
.as_ref()
.map(|trusted| trusted.evidence.clone()),
trusted_full_author_count: subject.trusted_full_author_count,
crawl_policy_max_follow_distance: subject.crawl_policy_max_follow_distance,
audit_mode: subject.scope.audit_mode().to_string(),
cutover_eligible: subject.scope == AuditScope::FullPolicyCutover
&& trusted_profile_rank_decisions.is_some()
&& subject.expected_profile_distance_seal_sha256.is_some()
&& stage_corpus.is_some(),
pool_catalog_sha256: catalog_before.sha256,
pool_manifest_sha256: catalog_before.manifest_sha256,
pool_stored_locations: catalog_before.stored_locations,
authors_processed: subject.authors_processed,
authors_total: subject.authors_total,
recovery_tranche_only: subject.scope.recovery_tranche_only(),
stage_corpus,
indexes,
profile,
queries,
representative_blocks,
};
let (_, full_cutover) = write_audit_output_noreplace(&output, prepared_output)?;
Ok(BulkProjectionAuditCompletion {
output,
full_cutover,
})
}
#[cfg(test)]
mod tests {
use super::*;
use hashtree_config::StorageBackend;
use hashtree_nostr::{stored_event_from_nostr_sdk_event, NostrEventStoreOptions};
use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
use super::super::super::{
persist_stage_state, run_nostr_bulk_projection_audit, CrawlStateLock,
IndexedNostrCrawlPolicy, StagedNostrCrawlState, STAGE_FORMAT_VERSION,
};
fn policy(author_count: usize) -> IndexedNostrCrawlPolicy {
IndexedNostrCrawlPolicy {
base_root: None,
author_allowlist_sha256: "ab".repeat(32),
author_count,
relays: vec!["wss://relay.example".to_string()],
require_all_relays: false,
max_events_seen: None,
max_authors: author_count,
max_follow_distance: Some(0),
max_live_bytes: 1_000_000,
author_batch_size: 1,
checkpoint_authors: 1,
per_author_event_limit: 256,
per_author_kind_event_limit: None,
per_author_live_bytes: Some(64 * 1024 * 1024),
fetch_timeout_millis: 30_000,
relay_event_max_bytes: Some(1024 * 1024),
global_relay_scan: false,
full_author_history: true,
negentropy_only: false,
relay_page_size: 1_000,
max_relay_pages: 67,
kinds: Some(vec![0, 1, 30_000]),
}
}
fn synthetic_root(byte: u8) -> String {
cid_to_nhash(&Cid::public([byte; 32])).expect("encode generated audit root")
}
fn full_cutover_output() -> BulkProjectionAuditOutput {
let indexes = NostrEventIndex::ALL
.into_iter()
.enumerate()
.map(|(position, index)| BulkProjectionIndexAudit {
index: index.name().to_string(),
root: Some(synthetic_root(position as u8 + 1)),
nodes: 1,
links: 1,
durable_values_validated: 1,
entries_sha256: "1".repeat(64),
retained_set_sha256: "2".repeat(64),
first_key: Some("first".to_string()),
last_key: Some("last".to_string()),
})
.collect();
let distance_seal = "3".repeat(64);
BulkProjectionAuditOutput {
version: BULK_PROJECTION_AUDIT_FORMAT_VERSION,
subject_kind: AuditSubjectKind::V3,
subject_version: 3,
candidate_root: synthetic_root(20),
state_sha256: "4".repeat(64),
stage_state_sha256: "5".repeat(64),
trusted_policy_sha256: "6".repeat(64),
policy_author_allowlist_sha256: "c".repeat(64),
trusted_profile_distance_seal_sha256: Some(distance_seal.clone()),
profile_distance_provenance: Some(ProfileDistanceProvenance {
format: PROFILE_RANK_DECISION_REPORT_FORMAT.to_string(),
census_format: PROFILE_RANK_DECISION_CENSUS_FORMAT.to_string(),
rank_decisions_file_sha256: "7".repeat(64),
rank_decisions_report_sha256: "8".repeat(64),
rank_decisions_sha256: "9".repeat(64),
social_graph_root: "a".repeat(64),
social_graph_sha256: "b".repeat(64),
eligible_authors_sha256: "c".repeat(64),
record_count: 1,
eligible_count: 1,
excluded_count: 0,
overmute_threshold: PROFILE_OVERMUTE_THRESHOLD,
census_max_distance: Some(4),
rank_policy: PROFILE_RANK_POLICY.to_string(),
exclusion_policy: PROFILE_EXCLUSION_POLICY.to_string(),
}),
trusted_full_author_count: 1,
crawl_policy_max_follow_distance: Some(0),
audit_mode: FULL_POLICY_CUTOVER_AUDIT_MODE.to_string(),
cutover_eligible: true,
pool_catalog_sha256: "d".repeat(64),
pool_manifest_sha256: "e".repeat(64),
pool_stored_locations: 1,
authors_processed: 1,
authors_total: 1,
recovery_tranche_only: false,
stage_corpus: Some(BulkProjectionStageCorpusAudit {
segment_count: 1,
event_cids_validated: 1,
retained_records: 1,
retained_records_sha256: "f".repeat(64),
replaceable_slots: 0,
replaceable_slots_sha256: "0".repeat(64),
}),
indexes,
profile: BulkProjectionProfileAudit {
by_pubkey_root: synthetic_root(21),
by_pubkey_root_file_sha256: "1".repeat(64),
by_pubkey_nodes: 1,
by_pubkey_links: 1,
by_pubkey_entries_sha256: "2".repeat(64),
search_root: synthetic_root(22),
search_root_file_sha256: "3".repeat(64),
search_nodes: 1,
search_entries: 1,
search_entries_sha256: "4".repeat(64),
sample_pubkey: "5".repeat(64),
sample_event_id: "6".repeat(64),
sample_name: "Generated profile".to_string(),
follow_distance_binding: "generated-test-binding".to_string(),
follow_distance_seal_sha256: distance_seal,
},
queries: vec![BulkProjectionQueryAudit {
query: "by-id".to_string(),
parameters: serde_json::json!({"limit": 1}),
event_ids: vec!["7".repeat(64)],
}],
representative_blocks: vec![BulkProjectionBlockEvidence {
role: "manifest".to_string(),
nhash: synthetic_root(20),
sha256: "8".repeat(64),
}],
}
}
fn assert_cutover_token_rejected(output: &BulkProjectionAuditOutput) {
let bytes = canonical_audit_output_bytes(output).expect("encode mutated audit evidence");
assert!(
mint_full_cutover_verified(output, &bytes).is_err(),
"mutated full-cutover evidence unexpectedly minted a token"
);
}
fn expected_profile_search_entry() -> (
ExpectedProfileSearchEntry,
hashtree_cli::socialgraph::StoredProfileSearchEntry,
) {
let keys = Keys::generate();
let event = EventBuilder::new(
Kind::Metadata,
r#"{"name":"Alice","display_name":"Alice Example","nip05":"alice@example.com"}"#,
)
.custom_created_at(Timestamp::from_secs(10))
.sign_with_keys(&keys)
.unwrap();
let mirrored_cid = Cid {
hash: [7; 32],
key: None,
};
let stored = hashtree_cli::socialgraph::stored_profile_search_entry_for_event(
&event,
&mirrored_cid,
Some(1),
)
.unwrap();
(
ExpectedProfileSearchEntry {
pubkey: event.pubkey.to_hex(),
event,
mirrored_cid,
},
stored,
)
}
fn rank_decision_artifacts(
decisions: &BTreeMap<String, Option<u32>>,
) -> (Vec<u8>, Vec<u8>, String, String) {
let mut digest = Sha256::new();
digest.update(PROFILE_RANK_DECISION_FORMAT.as_bytes());
digest.update(b"\n");
for (pubkey, rank) in decisions {
let row = match rank {
Some(rank) => serde_json::json!([pubkey, "eligible", rank]),
None => serde_json::json!([pubkey, "excluded", null]),
};
digest.update(serde_json::to_string(&row).unwrap().as_bytes());
digest.update(b"\n");
}
let decisions_sha256 = hex::encode(digest.finalize());
let mut lines = vec![serde_json::to_string(&ProfileRankDecisionHeader {
format: PROFILE_RANK_DECISION_FORMAT.to_string(),
eligible_ranks_sha256: decisions_sha256.clone(),
record_count: decisions.len(),
})
.unwrap()];
lines.extend(decisions.iter().map(|(pubkey, rank)| {
serde_json::to_string(&ProfileRankDecisionRecord {
pubkey: pubkey.clone(),
decision: if rank.is_some() {
"eligible"
} else {
"excluded"
}
.to_string(),
rank_hint: *rank,
})
.unwrap()
}));
let decisions_bytes = format!("{}\n", lines.join("\n")).into_bytes();
let decisions_file_sha256 = bytes_sha256(&decisions_bytes);
let eligible_count = decisions.values().filter(|rank| rank.is_some()).count();
let mut eligible_authors = decisions
.iter()
.filter_map(|(pubkey, rank)| rank.map(|rank| (rank, pubkey)))
.collect::<Vec<_>>();
eligible_authors.sort_unstable();
let eligible_authors_text = eligible_authors
.iter()
.map(|(_, pubkey)| format!("{pubkey}\n"))
.collect::<String>();
let eligible_authors_sha256 = bytes_sha256(eligible_authors_text.as_bytes());
let report = serde_json::json!({
"format": PROFILE_RANK_DECISION_REPORT_FORMAT,
"censusFormat": PROFILE_RANK_DECISION_CENSUS_FORMAT,
"socialGraphRoot": "a".repeat(64),
"socialGraphSha256": "b".repeat(64),
"eligibleAuthorsSha256": eligible_authors_sha256,
"overmuteThreshold": PROFILE_OVERMUTE_THRESHOLD,
"maxDistance": 4,
"rankPolicy": PROFILE_RANK_POLICY,
"exclusionPolicy": PROFILE_EXCLUSION_POLICY,
"recordCount": decisions.len(),
"eligibleCount": eligible_count,
"excludedCount": decisions.len() - eligible_count,
"reachableCount": eligible_count,
"reachableOvermutedCount": 0,
"distanceExcludedCount": 0,
"unreachableCount": decisions.len() - eligible_count,
"allGraphOvermutedCount": 0,
"rankDecisionsSha256": decisions_sha256,
"rankDecisionsFileSha256": decisions_file_sha256,
});
let report_bytes =
format!("{}\n", serde_json::to_string_pretty(&report).unwrap()).into_bytes();
let report_sha256 = bytes_sha256(&report_bytes);
(
decisions_bytes,
report_bytes,
decisions_file_sha256,
report_sha256,
)
}
#[test]
fn loads_pinned_rank_decision_provenance_and_rechecks_exact_bytes() {
let temp = tempfile::tempdir().unwrap();
let pubkey = "d".repeat(64);
let decisions = BTreeMap::from([(pubkey.clone(), Some(1))]);
let (decisions_bytes, report_bytes, decisions_sha256, report_sha256) =
rank_decision_artifacts(&decisions);
let decisions_path = temp.path().join("rank-decisions.jsonl");
let report_path = temp.path().join("rank-decisions-report.json");
std::fs::write(&decisions_path, decisions_bytes).unwrap();
std::fs::write(&report_path, &report_bytes).unwrap();
let options = BulkProjectionAuditOptions {
staging_data_dir: temp.path().join("unused-staging"),
expected_state_sha256: "1".repeat(64),
v3_candidate: false,
expected_stage_state_sha256: Some("2".repeat(64)),
expected_policy_sha256: Some("3".repeat(64)),
expected_profile_distance_seal_sha256: None,
profile_rank_decisions_file: Some(decisions_path.clone()),
expected_profile_rank_decisions_file_sha256: Some(decisions_sha256),
profile_rank_decisions_report: Some(report_path.clone()),
expected_profile_rank_decisions_report_sha256: Some(report_sha256),
expected_full_author_count: Some(1),
allow_recovery_tranche: false,
btree_order: Some(2),
page_size: 1,
query_limit: 1,
out: temp.path().join("unused-evidence.json"),
};
let trusted = load_trusted_profile_rank_decisions(&options)
.unwrap()
.expect("full provenance");
assert_eq!(trusted.decisions, decisions);
assert_eq!(trusted.evidence.eligible_count, 1);
assert_eq!(trusted.evidence.census_max_distance, Some(4));
let crawl_policy_max_follow_distance = Some(0);
assert_ne!(
trusted.evidence.census_max_distance,
crawl_policy_max_follow_distance
);
let eligible_authors_sha256 = eligible_profile_rank_authors_sha256(&decisions);
require_profile_rank_policy_binding(Some(&trusted), &eligible_authors_sha256, 1).unwrap();
assert!(
require_profile_rank_policy_binding(Some(&trusted), &"f".repeat(64), 1)
.unwrap_err()
.to_string()
.contains("does not match crawl-policy allowlist digest")
);
assert!(
require_profile_rank_policy_binding(Some(&trusted), &eligible_authors_sha256, 2)
.unwrap_err()
.to_string()
.contains("does not match crawl-policy author count")
);
recheck_trusted_profile_rank_decisions(Some(&trusted)).unwrap();
let alternate_report_path = temp.path().join("alternate-rank-report.json");
let mut alternate_report_bytes = report_bytes;
alternate_report_bytes.insert(alternate_report_bytes.len() - 1, b' ');
std::fs::write(&alternate_report_path, &alternate_report_bytes).unwrap();
let alternate_report_sha256 = bytes_sha256(&alternate_report_bytes);
let alternate = load_pinned_profile_rank_decisions(
&decisions_path,
&trusted.evidence.rank_decisions_file_sha256,
&alternate_report_path,
&alternate_report_sha256,
)
.expect("load semantically identical report with distinct exact bytes");
assert_eq!(alternate.decisions, trusted.decisions);
assert_ne!(
alternate.evidence.rank_decisions_report_sha256,
trusted.evidence.rank_decisions_report_sha256
);
assert!(
require_subject_profile_rank_authority(Some(&trusted.evidence), Some(&alternate),)
.expect_err(
"same decisions with a substituted report must not inherit v3 authority"
)
.to_string()
.contains("differs from the exact authority")
);
std::fs::write(&report_path, b"changed").unwrap();
assert!(recheck_trusted_profile_rank_decisions(Some(&trusted))
.unwrap_err()
.to_string()
.contains("changed during read-only audit"));
}
#[test]
fn pinned_rank_decisions_bind_the_producer_census_order_not_jsonl_key_order() {
let temp = tempfile::tempdir().unwrap();
let mut pubkeys = (0..4)
.map(|_| Keys::generate().public_key().to_hex())
.collect::<Vec<_>>();
pubkeys.sort();
let decisions = BTreeMap::from([
(pubkeys[0].clone(), Some(2)),
(pubkeys[1].clone(), None),
(pubkeys[2].clone(), Some(0)),
(pubkeys[3].clone(), Some(1)),
]);
let (decisions_bytes, report_bytes, decisions_sha256, report_sha256) =
rank_decision_artifacts(&decisions);
let decisions_path = temp.path().join("rank-decisions.jsonl");
let report_path = temp.path().join("rank-decisions-report.json");
std::fs::write(&decisions_path, decisions_bytes).unwrap();
std::fs::write(&report_path, &report_bytes).unwrap();
let census_order_text = format!("{}\n{}\n{}\n", pubkeys[2], pubkeys[3], pubkeys[0]);
let census_order_sha256 = bytes_sha256(census_order_text.as_bytes());
let pubkey_order_text = format!("{}\n{}\n{}\n", pubkeys[0], pubkeys[2], pubkeys[3]);
let pubkey_order_sha256 = bytes_sha256(pubkey_order_text.as_bytes());
assert_ne!(
census_order_sha256, pubkey_order_sha256,
"generated contract must distinguish census order from decision-key order"
);
let trusted = load_pinned_profile_rank_decisions(
&decisions_path,
&decisions_sha256,
&report_path,
&report_sha256,
)
.expect("accept generated producer-compatible census/report contract");
assert_eq!(
trusted.evidence.eligible_authors_sha256,
census_order_sha256
);
require_profile_rank_policy_binding(
Some(&trusted),
&census_order_sha256,
trusted.evidence.eligible_count,
)
.expect("bind the exact distance-ordered crawl allowlist");
let mut tampered_report: serde_json::Value = serde_json::from_slice(&report_bytes).unwrap();
tampered_report["eligibleAuthorsSha256"] = serde_json::Value::String(pubkey_order_sha256);
let tampered_report_bytes = format!(
"{}\n",
serde_json::to_string_pretty(&tampered_report).unwrap()
)
.into_bytes();
let tampered_report_path = temp.path().join("tampered-rank-decisions-report.json");
std::fs::write(&tampered_report_path, &tampered_report_bytes).unwrap();
let tampered_report_sha256 = bytes_sha256(&tampered_report_bytes);
let error = load_pinned_profile_rank_decisions(
&decisions_path,
&decisions_sha256,
&tampered_report_path,
&tampered_report_sha256,
)
.expect_err("pubkey-order digest must not inherit census authority");
assert!(error
.to_string()
.contains("does not match actual eligible decisions in census"));
}
#[test]
fn pinned_rank_decisions_reject_same_count_swapped_eligible_pubkey() {
let temp = tempfile::tempdir().unwrap();
let expected_pubkey = "d".repeat(64);
let expected = BTreeMap::from([(expected_pubkey, Some(1))]);
let expected_eligible_sha256 = eligible_profile_rank_authors_sha256(&expected);
let swapped_pubkey = "e".repeat(64);
let swapped = BTreeMap::from([(swapped_pubkey, Some(1))]);
let (decisions_bytes, report_bytes, decisions_sha256, _) =
rank_decision_artifacts(&swapped);
let mut report: serde_json::Value = serde_json::from_slice(&report_bytes).unwrap();
report["eligibleAuthorsSha256"] = serde_json::Value::String(expected_eligible_sha256);
let report_bytes =
format!("{}\n", serde_json::to_string_pretty(&report).unwrap()).into_bytes();
let report_sha256 = bytes_sha256(&report_bytes);
let decisions_path = temp.path().join("rank-decisions.jsonl");
let report_path = temp.path().join("rank-decisions-report.json");
std::fs::write(&decisions_path, decisions_bytes).unwrap();
std::fs::write(&report_path, report_bytes).unwrap();
let error = load_pinned_profile_rank_decisions(
&decisions_path,
&decisions_sha256,
&report_path,
&report_sha256,
)
.expect_err("same-count swapped eligible keys must not inherit the policy digest");
assert!(error
.to_string()
.contains("does not match actual eligible decisions in census"));
}
#[test]
fn v3_subject_adapter_rejects_unrelated_pinned_bytes_without_raw_authority_overrides() {
let temp = tempfile::tempdir().unwrap();
let data_dir = temp.path().join("data");
let state_path = data_dir
.join("nostr-index")
.join("bulk-projection-v3")
.join("state.json");
std::fs::create_dir_all(state_path.parent().unwrap()).unwrap();
let unrelated = b"{\"phase\":\"candidate\",\"root\":\"caller-chosen\"}\n";
std::fs::write(&state_path, unrelated).unwrap();
let spec = V3AuditSubjectSpec {
expected_state_sha256: bytes_sha256(unrelated),
staging_data_dir: temp.path().join("staging"),
};
let error = load_v3_audit_subject(&data_dir, spec)
.unwrap_err()
.to_string();
assert!(
error.contains("parse v3 state"),
"unrelated caller-pinned bytes must not become a v3 audit subject: {error}"
);
}
#[test]
fn only_complete_full_cutover_evidence_mints_typed_token() {
let output = full_cutover_output();
let bytes = canonical_audit_output_bytes(&output).unwrap();
let token = mint_full_cutover_verified(&output, &bytes)
.unwrap()
.expect("complete full-cutover evidence must mint a token");
assert_eq!(token.evidence_sha256, bytes_sha256(&bytes));
assert_eq!(token.subject_kind, AuditSubjectKind::V3);
assert_eq!(token.index_roots.len(), NostrEventIndex::ALL.len());
let mut legacy_v2_subject = output.clone();
legacy_v2_subject.subject_kind = AuditSubjectKind::V2;
legacy_v2_subject.subject_version = BULK_PROJECTION_VERSION;
assert_cutover_token_rejected(&legacy_v2_subject);
let mut recovery = output.clone();
recovery.audit_mode = RECOVERY_TRANCHE_AUDIT_MODE.to_string();
recovery.cutover_eligible = false;
recovery.recovery_tranche_only = true;
recovery.authors_processed = 1;
recovery.authors_total = 2;
recovery.trusted_full_author_count = 2;
recovery.stage_corpus = None;
let recovery_bytes = canonical_audit_output_bytes(&recovery).unwrap();
assert!(mint_full_cutover_verified(&recovery, &recovery_bytes)
.unwrap()
.is_none());
let mut wrong_mode = output.clone();
wrong_mode.audit_mode = RECOVERY_TRANCHE_AUDIT_MODE.to_string();
assert_cutover_token_rejected(&wrong_mode);
let mut not_eligible = output.clone();
not_eligible.cutover_eligible = false;
assert_cutover_token_rejected(¬_eligible);
let mut falsely_recovery = output.clone();
falsely_recovery.recovery_tranche_only = true;
assert_cutover_token_rejected(&falsely_recovery);
let mut missing_stage_corpus = output.clone();
missing_stage_corpus.stage_corpus = None;
assert_cutover_token_rejected(&missing_stage_corpus);
let mut empty_stage_corpus = output.clone();
empty_stage_corpus
.stage_corpus
.as_mut()
.expect("generated stage-corpus evidence")
.event_cids_validated = 0;
assert_cutover_token_rejected(&empty_stage_corpus);
let mut malformed_stage_digest = output.clone();
malformed_stage_digest
.stage_corpus
.as_mut()
.expect("generated stage-corpus evidence")
.retained_records_sha256 = "not-a-digest".to_string();
assert_cutover_token_rejected(&malformed_stage_digest);
let mut partial_count = output.clone();
partial_count.authors_processed = 0;
assert_cutover_token_rejected(&partial_count);
let mut missing_rank = output.clone();
missing_rank.profile_distance_provenance = None;
assert_cutover_token_rejected(&missing_rank);
let mut wrong_allowlist = output.clone();
wrong_allowlist.policy_author_allowlist_sha256 = "f".repeat(64);
assert_cutover_token_rejected(&wrong_allowlist);
let mut unpinned_profile = output.clone();
unpinned_profile.trusted_profile_distance_seal_sha256 = None;
assert_cutover_token_rejected(&unpinned_profile);
let mut missing_index = output.clone();
missing_index.indexes.pop();
assert_cutover_token_rejected(&missing_index);
let mut empty_index = output.clone();
empty_index.indexes[0].root = None;
assert_cutover_token_rejected(&empty_index);
let mut empty_pool = output.clone();
empty_pool.pool_stored_locations = 0;
assert_cutover_token_rejected(&empty_pool);
let mut bad_profile = output.clone();
bad_profile.profile.search_root = "not-a-root".to_string();
assert_cutover_token_rejected(&bad_profile);
let mut no_queries = output.clone();
no_queries.queries.clear();
assert_cutover_token_rejected(&no_queries);
let mut no_blocks = output;
no_blocks.representative_blocks.clear();
assert_cutover_token_rejected(&no_blocks);
}
#[test]
fn profile_search_audit_rejects_corrupt_known_fields_and_unknown_fields() {
let (expected, stored) = expected_profile_search_entry();
for field in ["name", "aliases", "nip05"] {
let mut value = serde_json::to_value(&stored).unwrap();
let object = value.as_object_mut().unwrap();
match field {
"name" => object.insert(field.to_string(), serde_json::json!("Mallory")),
"aliases" => object.insert(field.to_string(), serde_json::json!(["Wrong Alias"])),
"nip05" => object.insert(field.to_string(), serde_json::json!("wrong@example.com")),
_ => unreachable!(),
};
let mut bound = BTreeMap::new();
assert!(
validate_profile_search_value(
"p:test:pubkey",
&serde_json::to_string(&value).unwrap(),
&expected,
&mut bound,
)
.unwrap_err()
.to_string()
.contains("builder reconstruction"),
"corrupt {field} must fail exact reconstruction"
);
}
let mut value = serde_json::to_value(&stored).unwrap();
value
.as_object_mut()
.unwrap()
.insert("unknown".to_string(), serde_json::json!(true));
let error = validate_profile_search_value(
"p:test:pubkey",
&serde_json::to_string(&value).unwrap(),
&expected,
&mut BTreeMap::new(),
)
.unwrap_err();
assert!(
error.to_string().contains("strictly decode"),
"unexpected unknown-field error: {error:#}"
);
}
#[test]
fn profile_search_audit_binds_one_historic_distance_across_all_keys() {
let (expected, first) = expected_profile_search_entry();
let mut bound = BTreeMap::new();
validate_profile_search_value(
"p:alice:pubkey",
&serde_json::to_string(&first).unwrap(),
&expected,
&mut bound,
)
.unwrap();
let second = hashtree_cli::socialgraph::stored_profile_search_entry_for_event(
&expected.event,
&expected.mirrored_cid,
Some(2),
)
.unwrap();
let error = validate_profile_search_value(
"p:example:pubkey",
&serde_json::to_string(&second).unwrap(),
&expected,
&mut bound,
)
.unwrap_err();
assert!(
error.to_string().contains("builder reconstruction"),
"unexpected distance mismatch error: {error:#}"
);
let mut consistently_wrong = BTreeMap::new();
for key in ["p:alice:pubkey", "p:example:pubkey"] {
validate_profile_search_value(
key,
&serde_json::to_string(&second).unwrap(),
&expected,
&mut consistently_wrong,
)
.unwrap();
}
let wrong_distances = consistently_wrong
.iter()
.map(|(pubkey, entry)| (pubkey.clone(), entry.follow_distance))
.collect();
let wrong_seal =
hashtree_cli::socialgraph::profile_follow_distance_seal_v2(&wrong_distances);
let correct_values = BTreeMap::from([(expected.pubkey.clone(), first)]);
let correct_distances = correct_values
.iter()
.map(|(pubkey, entry)| (pubkey.clone(), entry.follow_distance))
.collect();
let correct_seal =
hashtree_cli::socialgraph::profile_follow_distance_seal_v2(&correct_distances);
assert_ne!(wrong_seal, correct_seal);
let trusted_decisions = BTreeMap::from([(expected.pubkey.clone(), Some(1))]);
let provenance_error =
validate_bound_profile_distances(&consistently_wrong, &trusted_decisions)
.expect_err("trusted rank decision must reject consistently wrong distance");
assert!(
provenance_error
.to_string()
.contains("trusted rank decision requires 1"),
"unexpected provenance mismatch: {provenance_error:#}"
);
let error =
require_profile_distance_attestation(false, Some(&correct_seal), &wrong_seal, true)
.expect_err("full cutover must reject consistently wrong historic distance");
assert!(
error.to_string().contains("SHA-256 mismatch"),
"unexpected full-cutover distance seal error: {error:#}"
);
}
#[test]
fn audit_output_refuses_live_trees_existing_targets_and_race_overwrites() {
let temp = tempfile::tempdir().unwrap();
let data_dir = temp.path().join("data");
let staging_dir = temp.path().join("staging");
let evidence_dir = temp.path().join("evidence");
std::fs::create_dir_all(&data_dir).unwrap();
std::fs::create_dir_all(&staging_dir).unwrap();
std::fs::create_dir_all(&evidence_dir).unwrap();
let protected = data_dir.join("audit.json");
assert!(
prepare_audit_output_path(&data_dir, &staging_dir, &protected)
.unwrap_err()
.to_string()
.contains("inside an audited live data tree")
);
let existing = evidence_dir.join("existing.json");
std::fs::write(&existing, b"trusted-existing").unwrap();
assert!(
prepare_audit_output_path(&data_dir, &staging_dir, &existing)
.unwrap_err()
.to_string()
.contains("already exists")
);
#[cfg(unix)]
{
let symlink = evidence_dir.join("symlink.json");
std::os::unix::fs::symlink(&existing, &symlink).unwrap();
assert!(prepare_audit_output_path(&data_dir, &staging_dir, &symlink)
.unwrap_err()
.to_string()
.contains("already exists"));
}
let raced = evidence_dir.join("raced.json");
let prepared = prepare_audit_output_path(&data_dir, &staging_dir, &raced).unwrap();
std::fs::write(&raced, b"racing-writer").unwrap();
assert!(install_audit_evidence_noreplace(b"new-evidence", &prepared).is_err());
assert_eq!(std::fs::read(&raced).unwrap(), b"racing-writer");
assert!(
std::fs::read_dir(&evidence_dir)
.unwrap()
.filter_map(|entry| entry.ok())
.all(|entry| !entry
.file_name()
.to_string_lossy()
.starts_with(".bulk-projection-audit.")),
"failed no-replace write left a partial temp file"
);
}
#[test]
fn audit_output_installs_complete_bytes_once_without_temp_files() {
let temp = tempfile::tempdir().unwrap();
let data_dir = temp.path().join("data");
let staging_dir = temp.path().join("staging");
let evidence_dir = temp.path().join("evidence");
std::fs::create_dir_all(&data_dir).unwrap();
std::fs::create_dir_all(&staging_dir).unwrap();
std::fs::create_dir_all(&evidence_dir).unwrap();
let target = evidence_dir.join("audit.json");
let prepared = prepare_audit_output_path(&data_dir, &staging_dir, &target).unwrap();
install_audit_evidence_noreplace(b"complete-evidence", &prepared).unwrap();
assert_eq!(std::fs::read(&target).unwrap(), b"complete-evidence");
assert_eq!(std::fs::read_dir(&evidence_dir).unwrap().count(), 1);
assert!(install_audit_evidence_noreplace(b"replacement", &prepared).is_err());
assert_eq!(std::fs::read(&target).unwrap(), b"complete-evidence");
}
#[tokio::test]
async fn audits_real_pool_spool_manifest_profiles_and_queries_read_only() {
let temp = tempfile::tempdir().unwrap();
let data_dir = temp.path().join("projection");
let staging_data_dir = temp.path().join("staging");
let evidence_path = temp.path().join("audit-evidence.json");
let store = hashtree_cli::HashtreeStore::with_options_and_backend(
&data_dir,
None,
0,
false,
&StorageBackend::Lmdb,
)
.unwrap();
let graph = hashtree_cli::socialgraph::open_social_graph_store_with_storage(
&data_dir,
store.store_arc(),
Some(128 * 1024 * 1024),
)
.unwrap();
let keys = Keys::generate();
let profile = EventBuilder::new(Kind::Metadata, r#"{"name":"Audit Alice"}"#)
.custom_created_at(Timestamp::from_secs(10))
.sign_with_keys(&keys)
.unwrap();
let note = EventBuilder::new(Kind::TextNote, "real audit note")
.tags([Tag::parse(["t", "hashtree"]).unwrap()])
.custom_created_at(Timestamp::from_secs(20))
.sign_with_keys(&keys)
.unwrap();
let parameterized = EventBuilder::new(Kind::Custom(30_000), "real parameterized event")
.tags([Tag::identifier("audit-article")])
.custom_created_at(Timestamp::from_secs(30))
.sign_with_keys(&keys)
.unwrap();
let parameterized_nul =
EventBuilder::new(Kind::Custom(30_000), "real NUL-key parameterized event")
.tags([Tag::identifier("audit-article\0suffix")])
.custom_created_at(Timestamp::from_secs(31))
.sign_with_keys(&keys)
.unwrap();
let stored = [&profile, ¬e, ¶meterized, ¶meterized_nul]
.into_iter()
.map(stored_event_from_nostr_sdk_event)
.collect::<Vec<_>>();
let event_store = NostrEventStore::with_options(
store.store_arc(),
NostrEventStoreOptions {
btree_order: Some(8),
..NostrEventStoreOptions::default()
},
);
let event_cids = event_store.store_event_blobs(stored.clone()).await.unwrap();
store.force_sync().unwrap();
let (state_path, spool_path) = bulk_paths(&data_dir);
let spool = BulkProjectionSpool::open(&spool_path).unwrap();
spool
.apply(stored.clone().into_iter().zip(event_cids).collect())
.unwrap();
graph
.sync_profile_index_for_events(std::slice::from_ref(&profile))
.unwrap();
graph.force_sync().unwrap();
let mut roots = BTreeMap::new();
for index in NostrEventIndex::ALL {
roots.insert(
index,
spool
.build_index_root(index, store.store_arc(), 8)
.await
.unwrap(),
);
}
let candidate_root = event_store
.write_bulk_index_manifest(&roots)
.await
.unwrap()
.unwrap();
store.force_sync().unwrap();
let policy = policy(101_267);
let allowlist_sha256 = policy.author_allowlist_sha256.clone();
let state = BulkProjectionState {
version: BULK_PROJECTION_VERSION,
author_allowlist_source: Some(format!(
"http://127.0.0.1:45442/eligible/{allowlist_sha256}"
)),
policy: policy.clone(),
next_author: 17_177,
segment_event_offset: 0,
events_seen: 4,
events_selected: 4,
live_bytes_selected: 123,
built_roots: roots
.iter()
.map(|(index, root)| {
(
index.stable_id(),
root.as_ref()
.map(cid_to_nhash)
.transpose()
.unwrap()
.unwrap_or_default(),
)
})
.collect(),
complete_root: Some(cid_to_nhash(&candidate_root).unwrap()),
};
super::super::persist_bulk_state(&state_path, &state).unwrap();
let stage = StagedNostrCrawlState {
version: STAGE_FORMAT_VERSION,
author_allowlist_source: Some(format!(
"http://127.0.0.1:45441/eligible/{allowlist_sha256}"
)),
policy,
next_author: state.next_author,
events_seen: state.events_seen,
events_selected: state.events_selected,
live_bytes_selected: state.live_bytes_selected,
};
assert_ne!(
state.author_allowlist_source, stage.author_allowlist_source,
"the production-shaped audit must tolerate distinct ephemeral source ports"
);
assert_eq!(
state.policy, stage.policy,
"the canonical policy, including allowlist digest/count, remains authoritative"
);
persist_stage_state(&staging_data_dir, &stage).unwrap();
drop(CrawlStateLock::acquire(&data_dir).unwrap());
drop(CrawlStateLock::acquire_stage(&staging_data_dir).unwrap());
drop(event_store);
drop(graph);
let spool_closing = spool.env.clone().prepare_for_closing();
drop(spool);
spool_closing.wait();
drop(store);
let exact_spool = BulkProjectionSpool::open_read_only(&spool_path).unwrap();
let exact_store =
Arc::new(ReadOnlyPoolStore::open(data_dir.join(SHARED_BLOB_POOL_DIR_NAME)).unwrap());
let exact = audit_exact_event_index_parity(
&exact_spool,
Arc::clone(&exact_store),
&state.built_roots,
8,
1,
)
.await
.unwrap();
assert_eq!(exact.format, EXACT_INDEX_PARITY_FORMAT);
assert_eq!(exact.spool_event_records, 4);
assert_eq!(exact.indexes.len(), 9);
for (position, index) in NostrEventIndex::ALL.iter().copied().enumerate() {
assert_eq!(exact.indexes[position].index, index.name());
assert_eq!(exact.indexes[position].stable_id, index.stable_id());
assert_eq!(
exact.indexes[position].root,
state.built_roots[&index.stable_id()]
);
assert_eq!(
exact.indexes[position].durable_values_validated,
if index == NostrEventIndex::ById { 4 } else { 0 }
);
}
validate_exact_event_index_parity_evidence(&exact, &state.built_roots, 8).unwrap();
let mut substituted_roots = state.built_roots.clone();
substituted_roots.insert(1, substituted_roots[&0].clone());
let substitution_error = audit_exact_event_index_parity(
&exact_spool,
Arc::clone(&exact_store),
&substituted_roots,
8,
1,
)
.await
.expect_err("substituting one structurally valid root must fail exact spool parity");
assert!(
format!("{substitution_error:#}").contains("mismatch"),
"unexpected substituted-root error: {substitution_error:#}"
);
let exact_spool_closing = exact_spool.env.clone().prepare_for_closing();
drop(exact_spool);
exact_spool_closing.wait();
drop(exact_store);
let state_sha256 = bytes_sha256(&std::fs::read(&state_path).unwrap());
let stage_path = staging_data_dir.join(STAGE_DIR).join(STAGE_STATE_FILE);
let stage_sha256 = bytes_sha256(&std::fs::read(stage_path).unwrap());
let policy_sha256 = bytes_sha256(&serde_json::to_vec(&state.policy).unwrap());
let full_mode_error = run_nostr_bulk_projection_audit(
data_dir.clone(),
BulkProjectionAuditOptions {
staging_data_dir: staging_data_dir.clone(),
expected_state_sha256: state_sha256.clone(),
v3_candidate: false,
expected_stage_state_sha256: Some(stage_sha256.clone()),
expected_policy_sha256: Some(policy_sha256.clone()),
expected_profile_distance_seal_sha256: None,
profile_rank_decisions_file: None,
expected_profile_rank_decisions_file_sha256: None,
profile_rank_decisions_report: None,
expected_profile_rank_decisions_report_sha256: None,
expected_full_author_count: Some(101_267),
allow_recovery_tranche: false,
btree_order: Some(8),
page_size: 2,
query_limit: 2,
out: temp.path().join("must-not-write-full-evidence.json"),
},
)
.await
.unwrap_err();
assert!(
full_mode_error.to_string().contains("full-policy audit"),
"unexpected full-mode error: {full_mode_error:#}"
);
let count_error = run_nostr_bulk_projection_audit(
data_dir.clone(),
BulkProjectionAuditOptions {
staging_data_dir: staging_data_dir.clone(),
expected_state_sha256: state_sha256.clone(),
v3_candidate: false,
expected_stage_state_sha256: Some(stage_sha256.clone()),
expected_policy_sha256: Some(policy_sha256.clone()),
expected_profile_distance_seal_sha256: None,
profile_rank_decisions_file: None,
expected_profile_rank_decisions_file_sha256: None,
profile_rank_decisions_report: None,
expected_profile_rank_decisions_report_sha256: None,
expected_full_author_count: Some(101_268),
allow_recovery_tranche: true,
btree_order: Some(8),
page_size: 2,
query_limit: 2,
out: temp.path().join("must-not-write-count-evidence.json"),
},
)
.await
.unwrap_err();
assert!(
count_error
.to_string()
.contains("policy author count mismatch"),
"unexpected trusted-count error: {count_error:#}"
);
let recovery_output = audit_bulk_projection(
&data_dir,
BulkProjectionAuditOptions {
staging_data_dir,
expected_state_sha256: state_sha256,
v3_candidate: false,
expected_stage_state_sha256: Some(stage_sha256),
expected_policy_sha256: Some(policy_sha256.clone()),
expected_profile_distance_seal_sha256: None,
profile_rank_decisions_file: None,
expected_profile_rank_decisions_file_sha256: None,
profile_rank_decisions_report: None,
expected_profile_rank_decisions_report_sha256: None,
expected_full_author_count: Some(101_267),
allow_recovery_tranche: true,
btree_order: Some(8),
page_size: 2,
query_limit: 2,
out: evidence_path.clone(),
},
)
.await
.unwrap();
let evidence_bytes = std::fs::read(evidence_path).unwrap();
assert!(
mint_full_cutover_verified(&recovery_output, &evidence_bytes)
.unwrap()
.is_none()
);
let output: serde_json::Value = serde_json::from_slice(&evidence_bytes).unwrap();
assert_eq!(output["version"], BULK_PROJECTION_AUDIT_FORMAT_VERSION);
assert_eq!(output["subject_kind"], "v2");
assert_eq!(output["subject_version"], BULK_PROJECTION_VERSION);
assert_eq!(
output["candidate_root"],
cid_to_nhash(&candidate_root).unwrap()
);
assert_eq!(output["recovery_tranche_only"], true);
assert_eq!(
output["audit_mode"],
"recovery-tranche-internal-non-cutover"
);
assert_eq!(output["cutover_eligible"], false);
assert_eq!(output["trusted_policy_sha256"], policy_sha256);
assert_eq!(
output["policy_author_allowlist_sha256"],
state.policy.author_allowlist_sha256
);
assert_eq!(output["trusted_full_author_count"], 101_267);
assert_eq!(output["crawl_policy_max_follow_distance"], 0);
assert_eq!(output["authors_processed"], 17_177);
assert_eq!(output["authors_total"], 101_267);
assert_eq!(output["indexes"].as_array().unwrap().len(), 9);
assert_eq!(output["indexes"][0]["durable_values_validated"], 4);
assert_eq!(output["profile"]["by_pubkey_links"], 1);
assert!(output["profile"]["search_entries"].as_u64().unwrap() >= 1);
assert_eq!(
output["profile"]["by_pubkey_root_file_sha256"]
.as_str()
.unwrap()
.len(),
64
);
assert_eq!(
output["profile"]["search_root_file_sha256"]
.as_str()
.unwrap()
.len(),
64
);
let query_names = output["queries"]
.as_array()
.unwrap()
.iter()
.map(|query| query["query"].as_str().unwrap())
.collect::<Vec<_>>();
assert!(query_names.contains(&"replaceable"));
assert!(query_names.contains(&"parameterized-replaceable"));
assert!(!output["representative_blocks"]
.as_array()
.unwrap()
.is_empty());
}
}