1use std::io;
8use std::path::Path;
9
10use serde::Serialize;
11
12use super::super::{
13 Q42Volume, FLAG_FIELD_POSTINGS, FLAG_FIELD_RANGES, FLAG_OBJECT_SORTED,
14};
15use super::manifest::Q42VolumeSet;
16use super::postings::validate_postings_section;
17
18#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
19#[serde(rename_all = "snake_case")]
20pub enum VerifyLevel {
21 Structure,
22 Blocks,
23 Lexicon,
24 Indexes,
25 Full,
26}
27
28impl VerifyLevel {
29 pub fn parse(raw: &str) -> Result<Self, String> {
30 match raw.trim().to_ascii_lowercase().as_str() {
31 "structure" => Ok(Self::Structure),
32 "blocks" => Ok(Self::Blocks),
33 "lexicon" => Ok(Self::Lexicon),
34 "indexes" => Ok(Self::Indexes),
35 "full" => Ok(Self::Full),
36 other => Err(format!("unknown verify level '{other}'")),
37 }
38 }
39}
40
41#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
42#[serde(rename_all = "snake_case")]
43pub enum CheckStatus {
44 Pass,
45 Fail,
46 NotApplicable,
47 NotChecked,
48 Incomplete,
49}
50
51#[derive(Clone, Debug, Serialize)]
52pub struct VerifyCheck {
53 pub name: String,
54 pub status: CheckStatus,
55 pub detail: String,
56}
57
58#[derive(Clone, Debug, Serialize)]
59pub struct Q42VerifyReceipt {
60 pub path: String,
61 pub level: VerifyLevel,
62 pub overall: CheckStatus,
63 pub checks: Vec<VerifyCheck>,
64}
65
66#[derive(Clone, Copy, Debug, Eq, PartialEq)]
67enum FileRole {
68 Standalone,
69 VolumeRoot,
70 DataChild { shared_lexicon: bool },
71 LexiconShard,
72}
73
74impl Q42VerifyReceipt {
75 pub fn from_path(path: &Path, level: VerifyLevel) -> io::Result<Self> {
76 let volume = Q42Volume::open(path)?;
77 Ok(Self::from_volume(path, &volume, level))
78 }
79
80 pub fn from_volume(path: &Path, volume: &Q42Volume, level: VerifyLevel) -> Self {
81 let role = if volume.volume_manifest().ok().flatten().is_some() {
82 FileRole::VolumeRoot
83 } else {
84 FileRole::Standalone
85 };
86 Self::from_volume_role(path, volume, level, role)
87 }
88
89 fn from_volume_role(
90 path: &Path,
91 volume: &Q42Volume,
92 level: VerifyLevel,
93 role: FileRole,
94 ) -> Self {
95 let mut checks = Vec::new();
96 let run_structure = matches!(
97 level,
98 VerifyLevel::Structure | VerifyLevel::Full | VerifyLevel::Blocks | VerifyLevel::Indexes | VerifyLevel::Lexicon
99 );
100 let run_blocks = matches!(level, VerifyLevel::Blocks | VerifyLevel::Full);
101 let run_lex = matches!(level, VerifyLevel::Lexicon | VerifyLevel::Full);
102 let run_idx = matches!(level, VerifyLevel::Indexes | VerifyLevel::Full);
103
104 if run_structure {
105 checks.push(check("structure.open", CheckStatus::Pass, "header and section bounds accepted on open"));
106 let flags = volume.header().flags;
107 if volume.block_count() > 0 && flags & FLAG_OBJECT_SORTED == 0 {
108 checks.push(check(
109 "structure.object_sorted",
110 CheckStatus::Fail,
111 "blocks present but object-sorted flag is clear",
112 ));
113 } else {
114 checks.push(check(
115 "structure.object_sorted",
116 CheckStatus::Pass,
117 "object sort flag matches block presence",
118 ));
119 }
120 }
121
122 if run_blocks {
123 if volume.volume_manifest().ok().flatten().is_some() {
124 checks.push(check(
125 "blocks.decode",
126 CheckStatus::NotApplicable,
127 "volume root has no local SuperBlocks; verify children",
128 ));
129 } else {
130 match volume.verify_all_blocks() {
131 Ok(receipt) => checks.push(check(
132 "blocks.decode",
133 CheckStatus::Pass,
134 format!(
135 "{} blocks, {} Quins, parity and object order ok",
136 receipt.blocks_verified, receipt.quins_verified
137 ),
138 )),
139 Err(error) => checks.push(check(
140 "blocks.decode",
141 CheckStatus::Fail,
142 error.to_string(),
143 )),
144 }
145 }
146 } else if level == VerifyLevel::Full {
147 checks.push(check(
148 "blocks.decode",
149 CheckStatus::NotChecked,
150 "block decode was not requested",
151 ));
152 }
153
154 if run_lex {
155 match volume.lex_view() {
156 Ok(lex) => {
157 let entries = lex.entry_count();
158 let (status, detail) = match role {
159 FileRole::DataChild {
160 shared_lexicon: true,
161 } if entries == 0 => (
162 CheckStatus::Pass,
163 "0 local terms; this data child uses the volume-set lexicon shards"
164 .into(),
165 ),
166 FileRole::VolumeRoot if entries == 0 => (
167 CheckStatus::Pass,
168 "root catalog has no local terms; lexicon shards are verified separately"
169 .into(),
170 ),
171 FileRole::LexiconShard if entries == 0 => (
172 CheckStatus::Fail,
173 "lexicon shard contains 0 terms".into(),
174 ),
175 FileRole::Standalone | FileRole::DataChild { shared_lexicon: false }
176 if entries == 0 && volume.block_count() > 0 =>
177 {
178 (
179 CheckStatus::Incomplete,
180 "data file has 0 embedded terms and no shared lexicon shards"
181 .into(),
182 )
183 }
184 _ => (
185 CheckStatus::Pass,
186 format!("{entries} recoverable terms in embedded Q42LEX"),
187 ),
188 };
189 checks.push(check("lexicon.entries", status, detail));
190 }
191 Err(error) => checks.push(check(
192 "lexicon.entries",
193 CheckStatus::Fail,
194 format!("{error:?}"),
195 )),
196 }
197 }
198
199 if run_structure {
200 let merkle = volume.header().merkle_root;
201 let dag_len = volume.header().dag_root_length;
202 let merkle_empty = merkle.iter().all(|b| *b == 0) && dag_len == 0;
203 let (status, detail) = match (role, merkle_empty) {
204 (FileRole::Standalone, true) if volume.block_count() > 0 => (
205 CheckStatus::Incomplete,
206 "merkle_root and DAG are empty on a standalone data file",
207 ),
208 (FileRole::DataChild { .. } | FileRole::VolumeRoot, true) => (
209 CheckStatus::NotApplicable,
210 "per-file DAG empty; volume-set identity is the root manifest SHA-256",
211 ),
212 (FileRole::LexiconShard, true) => (
213 CheckStatus::NotApplicable,
214 "lexicon shard has no SuperBlock DAG",
215 ),
216 (_, false) => (CheckStatus::Pass, "header merkle_root / DAG section present"),
217 _ => (CheckStatus::NotApplicable, "no data blocks"),
218 };
219 checks.push(check("structure.merkle", status, detail));
220 }
221
222 if run_idx {
223 let bidx = volume.bidx_bytes();
224 if bidx.is_empty() && volume.block_count() > 0 {
225 checks.push(check(
226 "indexes.bidx",
227 CheckStatus::Fail,
228 "blocks present but BIDX is empty",
229 ));
230 } else if bidx.is_empty() {
231 checks.push(check(
232 "indexes.bidx",
233 CheckStatus::NotApplicable,
234 "no blocks",
235 ));
236 } else {
237 match super::index::validate_bidx(bidx, volume.block_count() as usize) {
238 Ok(()) => checks.push(check("indexes.bidx", CheckStatus::Pass, "BIDX layout and monotonicity ok")),
239 Err(error) => checks.push(check("indexes.bidx", CheckStatus::Fail, error.to_string())),
240 }
241 }
242 let flags = volume.header().flags;
243 if flags & FLAG_FIELD_RANGES != 0 {
244 checks.push(check(
245 "indexes.field_ranges",
246 CheckStatus::Pass,
247 "FIDX present and accepted on open",
248 ));
249 } else if volume.block_count() > 0 && !matches!(role, FileRole::LexiconShard) {
250 checks.push(check(
251 "indexes.field_ranges",
252 CheckStatus::Incomplete,
253 "no FIDX on a data segment",
254 ));
255 }
256 if flags & FLAG_FIELD_POSTINGS != 0 {
257 if let Some((offset, length)) = volume.header().field_postings_range() {
258 let start = offset as usize;
259 let end = start + length as usize;
260 match volume.as_bytes().get(start..end) {
261 Some(bytes) => match validate_postings_section(bytes, volume.block_count() as usize) {
262 Ok(()) => checks.push(check(
263 "indexes.postings",
264 CheckStatus::Pass,
265 "PIDX layout ok",
266 )),
267 Err(error) => checks.push(check(
268 "indexes.postings",
269 CheckStatus::Fail,
270 error.to_string(),
271 )),
272 },
273 None => checks.push(check(
274 "indexes.postings",
275 CheckStatus::Fail,
276 "PIDX range outside file",
277 )),
278 }
279 }
280 } else if volume.block_count() > 0 && !matches!(role, FileRole::LexiconShard) {
281 checks.push(check(
282 "indexes.postings",
283 CheckStatus::Incomplete,
284 "no PIDX on a data segment",
285 ));
286 }
287 }
288
289 let overall = overall_status(level, &checks);
290 Self {
291 path: path.display().to_string(),
292 level,
293 overall,
294 checks,
295 }
296 }
297
298 pub fn to_text(&self) -> String {
299 let mut out = format!(
300 "Q42 verify {} level={:?} overall={:?}\n",
301 self.path, self.level, self.overall
302 );
303 for check in &self.checks {
304 out.push_str(&format!(
305 " {:<22} {:<14} {}\n",
306 check.name,
307 format!("{:?}", check.status),
308 check.detail
309 ));
310 }
311 out
312 }
313}
314
315fn check(name: &str, status: CheckStatus, detail: impl Into<String>) -> VerifyCheck {
316 VerifyCheck {
317 name: name.into(),
318 status,
319 detail: detail.into(),
320 }
321}
322
323fn overall_status(level: VerifyLevel, checks: &[VerifyCheck]) -> CheckStatus {
324 if checks.iter().any(|c| c.status == CheckStatus::Fail) {
325 return CheckStatus::Fail;
326 }
327 if level == VerifyLevel::Full {
328 if checks.iter().any(|c| {
329 matches!(
330 c.status,
331 CheckStatus::NotChecked | CheckStatus::Incomplete
332 )
333 }) {
334 return CheckStatus::Incomplete;
335 }
336 }
337 if checks.iter().any(|c| c.status == CheckStatus::Incomplete) {
338 return CheckStatus::Incomplete;
339 }
340 CheckStatus::Pass
341}
342
343#[derive(Clone, Debug, Serialize)]
345pub struct Q42VerifySetReport {
346 pub root: String,
347 pub overall: CheckStatus,
348 pub members: Vec<Q42VerifyReceipt>,
349}
350
351impl Q42VerifySetReport {
352 pub fn to_text(&self) -> String {
353 let mut out = format!(
354 "Q42 verify-set {} members={} overall={:?}\n",
355 self.root,
356 self.members.len(),
357 self.overall
358 );
359 for receipt in &self.members {
360 out.push_str(&receipt.to_text());
361 }
362 out
363 }
364}
365
366pub fn verify_volume_set_from_root(path: &Path, level: VerifyLevel) -> io::Result<Q42VerifySetReport> {
368 let root = Q42Volume::open(path)?;
369 let Some(manifest) = root.volume_manifest()? else {
370 let receipt = Q42VerifyReceipt::from_volume(path, &root, level);
371 return Ok(Q42VerifySetReport {
372 overall: receipt.overall,
373 root: path.display().to_string(),
374 members: vec![receipt],
375 });
376 };
377
378 let parent = path.parent().unwrap_or(Path::new("."));
379 let shared_lexicon = !manifest.lexicon_segments.is_empty();
380 let mut members = vec![Q42VerifyReceipt::from_volume_role(
381 path,
382 &root,
383 level,
384 FileRole::VolumeRoot,
385 )];
386
387 let mut digest_ok = true;
388 let digest_detail;
389 match Q42VolumeSet::open_root(path) {
390 Ok(set) => match set.verify_segment_hashes(path) {
391 Ok(()) => digest_detail = format!(
392 "{} data + {} lexicon shard digest(s) match the root manifest",
393 manifest.segments.len(),
394 manifest.lexicon_segments.len()
395 ),
396 Err(error) => {
397 digest_ok = false;
398 digest_detail = error.to_string();
399 }
400 },
401 Err(error) => {
402 digest_ok = false;
403 digest_detail = error.to_string();
404 }
405 }
406 members[0].checks.push(check(
407 "set.digests",
408 if digest_ok {
409 CheckStatus::Pass
410 } else {
411 CheckStatus::Fail
412 },
413 digest_detail,
414 ));
415 members[0].overall = overall_status(level, &members[0].checks);
416
417 for entry in &manifest.segments {
418 let child = parent.join(&entry.locator);
419 if !child.is_file() {
420 members.push(Q42VerifyReceipt {
421 path: child.display().to_string(),
422 level,
423 overall: CheckStatus::Fail,
424 checks: vec![check(
425 "set.member",
426 CheckStatus::Fail,
427 "data child is missing",
428 )],
429 });
430 continue;
431 }
432 let volume = Q42Volume::open(&child)?;
433 members.push(Q42VerifyReceipt::from_volume_role(
434 &child,
435 &volume,
436 level,
437 FileRole::DataChild { shared_lexicon },
438 ));
439 }
440 for entry in &manifest.lexicon_segments {
441 let child = parent.join(&entry.locator);
442 if !child.is_file() {
443 members.push(Q42VerifyReceipt {
444 path: child.display().to_string(),
445 level,
446 overall: CheckStatus::Fail,
447 checks: vec![check(
448 "set.member",
449 CheckStatus::Fail,
450 "lexicon shard is missing",
451 )],
452 });
453 continue;
454 }
455 let volume = Q42Volume::open(&child)?;
456 members.push(Q42VerifyReceipt::from_volume_role(
457 &child,
458 &volume,
459 level,
460 FileRole::LexiconShard,
461 ));
462 }
463
464 let overall = fold_set_overall(level, &members);
465 Ok(Q42VerifySetReport {
466 root: path.display().to_string(),
467 overall,
468 members,
469 })
470}
471
472fn fold_set_overall(level: VerifyLevel, members: &[Q42VerifyReceipt]) -> CheckStatus {
473 if members.iter().any(|m| m.overall == CheckStatus::Fail) {
474 return CheckStatus::Fail;
475 }
476 if members
477 .iter()
478 .any(|m| m.overall == CheckStatus::Incomplete)
479 || (level == VerifyLevel::Full
480 && members
481 .iter()
482 .any(|m| m.checks.iter().any(|c| c.status == CheckStatus::NotChecked)))
483 {
484 return CheckStatus::Incomplete;
485 }
486 CheckStatus::Pass
487}
488
489#[cfg(test)]
490mod tests {
491 use super::*;
492 use crate::q42_volume::{write_unified_volume, StreamingQ42VolumeWriter};
493 use crate::NQuin;
494 use std::collections::HashMap;
495
496 #[test]
497 fn full_on_hashed_only_file_is_incomplete_not_pass() {
498 let file = tempfile::NamedTempFile::new().unwrap();
499 write_unified_volume(
500 file.path(),
501 &HashMap::new(),
502 &[(3, 3)],
503 &[vec![NQuin {
504 subject: 1,
505 predicate: 2,
506 object: 3,
507 context: 0,
508 metadata: 0,
509 parity: 1 ^ 2 ^ 3,
510 }]],
511 )
512 .unwrap();
513 let receipt = Q42VerifyReceipt::from_path(file.path(), VerifyLevel::Full).unwrap();
514 assert_eq!(receipt.overall, CheckStatus::Incomplete);
515 assert!(receipt.checks.iter().any(|c| c.name == "lexicon.entries"));
516 }
517
518 #[test]
519 fn full_on_lex_and_postings_passes() {
520 let file = tempfile::NamedTempFile::new().unwrap();
521 let mut lex = HashMap::new();
522 lex.insert(1, "s".into());
523 lex.insert(2, "p".into());
524 lex.insert(3, "o".into());
525 let mut writer = StreamingQ42VolumeWriter::new(&lex).unwrap();
526 writer
527 .push_block(
528 0,
529 &[NQuin {
530 subject: 1,
531 predicate: 2,
532 object: 3,
533 context: 0,
534 metadata: 0,
535 parity: 1 ^ 2 ^ 3,
536 }],
537 )
538 .unwrap();
539 writer.finish(file.path()).unwrap();
540 let receipt = Q42VerifyReceipt::from_path(file.path(), VerifyLevel::Full).unwrap();
541 assert_eq!(receipt.overall, CheckStatus::Pass, "{:?}", receipt.checks);
542 }
543
544 #[test]
545 fn volume_set_full_passes_with_shared_lexicon() {
546 use crate::q42_volume::{
547 write_volume_root_for_commons, Q42VolumeManifest, StreamingQ42VolumeWriter,
548 };
549
550 let dir = tempfile::TempDir::new().unwrap();
551 let mut lex = HashMap::new();
552 lex.insert(1, "s".into());
553 lex.insert(2, "p".into());
554 lex.insert(3, "o".into());
555 let child = dir.path().join("child.q42");
556 let mut data = StreamingQ42VolumeWriter::new(&HashMap::new()).unwrap();
557 data.declare_permissive_commons();
558 data.push_block(
559 0,
560 &[NQuin {
561 subject: 1,
562 predicate: 2,
563 object: 3,
564 context: 0,
565 metadata: 0,
566 parity: 1 ^ 2 ^ 3,
567 }],
568 )
569 .unwrap();
570 data.finish(&child).unwrap();
571 let shard = dir.path().join("lex-00000.q42");
572 let mut words = StreamingQ42VolumeWriter::new(&lex).unwrap();
573 words.declare_permissive_commons();
574 words.finish(&shard).unwrap();
575 let root = dir.path().join("root.q42");
576 write_volume_root_for_commons(
577 &root,
578 &Q42VolumeManifest {
579 generation: 1,
580 segments: vec![
581 Q42VolumeManifest::segment_from_file(&child, "child.q42".into()).unwrap(),
582 ],
583 lexicon_segments: vec![Q42VolumeManifest::lexicon_segment_from_file(
584 &shard,
585 "lex-00000.q42".into(),
586 )
587 .unwrap()],
588 },
589 )
590 .unwrap();
591
592 let report = verify_volume_set_from_root(&root, VerifyLevel::Full).unwrap();
593 assert_eq!(report.members.len(), 3, "{:?}", report.members);
594 assert_eq!(report.overall, CheckStatus::Pass, "{}", report.to_text());
595 assert!(report
596 .members
597 .iter()
598 .any(|m| m.checks.iter().any(|c| c.name == "set.digests"
599 && c.status == CheckStatus::Pass)));
600 }
601}