diff --git a/.config/make/tests.mak b/.config/make/tests.mak index fb29185e3..b60d4be67 100644 --- a/.config/make/tests.mak +++ b/.config/make/tests.mak @@ -32,8 +32,11 @@ script-tests: ## Run shell script tests ./scripts/test_hotpath_warp_ab_gate.sh ./scripts/test_hotpath_warp_abba.sh ./scripts/test_scanner_validation_harness.sh + ./scripts/test_scanner_heal_checkpoint_crash_evidence.sh ./scripts/test_scanner_heal_g14_multiset_evidence.sh ./scripts/test_scanner_heal_scheduler_pressure_evidence.sh + ./scripts/test_scanner_heal_status_outcome_evidence.sh + ./scripts/test_scanner_heal_maintenance_evidence.sh ./scripts/test_scanner_heal_w13_mrf_evidence.sh ./scripts/test_scanner_heal_w16_recovery_evidence.sh ./scripts/test_exact_1mib_handoff_abba.sh diff --git a/.config/scanner-heal-required-tests.json b/.config/scanner-heal-required-tests.json index f80e5664a..694c9a5cc 100644 --- a/.config/scanner-heal-required-tests.json +++ b/.config/scanner-heal-required-tests.json @@ -209,7 +209,11 @@ "lane": "status-and-outcome", "status": "pending", "description": "Per-object outcomes and bounded terminal retention", - "requires": ["per-object outcome oracle", "terminal retention bounds"] + "requires": ["per-object outcome oracle", "terminal retention bounds"], + "evidence_fields": [ + "per_object_outcome_oracle", + "terminal_retention_bounds" + ] }, { "gate": "G06", @@ -217,7 +221,12 @@ "lane": "status-and-outcome", "status": "pending", "description": "Concurrent status, legacy clients and truncation", - "requires": ["concurrent status evidence", "legacy client compatibility", "truncation behavior"] + "requires": ["concurrent status evidence", "legacy client compatibility", "truncation behavior"], + "evidence_fields": [ + "concurrent_status_evidence", + "legacy_client_compatibility", + "truncation_behavior" + ] }, { "gate": "G07", @@ -375,7 +384,13 @@ "lane": "status-and-outcome", "status": "pending", "description": "Manager-to-event-to-ledger exact disposition, including grace", - "requires": ["manager disposition evidence", "event disposition evidence", "ledger disposition evidence", "grace handling"] + "requires": ["manager disposition evidence", "event disposition evidence", "ledger disposition evidence", "grace handling"], + "evidence_fields": [ + "manager_disposition_evidence", + "event_disposition_evidence", + "ledger_disposition_evidence", + "grace_handling" + ] }, { "gate": "R-L", diff --git a/scripts/README.md b/scripts/README.md index fa7eba564..82a4598b5 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -56,14 +56,20 @@ their issue closes. | `probe.sh` | dev-tool | Probe-style e2e run | `make probe-e2e` | | `run_scanner_validation_harness.sh` | dev-tool | Scanner validation harness | `docs/operations/scanner-benchmark-runbook.md` | | `run_scanner_heal_evidence_case.sh` | dev-tool | Runs one Scanner/Heal release-evidence registry case and checks the produced receipt/oracle | `.config/scanner-heal-required-tests.json`; `check_test_wiring.py --check-scanner-heal` | +| `run_scanner_heal_checkpoint_crash_evidence.py` | dev-tool | Assembles measured Scanner/Heal G02/R-E checkpoint and restart release descriptors from scanner restart diagnostic reports | `diagnose_scanner_enumeration_restart.py`; `test_scanner_heal_checkpoint_crash_evidence.sh` | | `run_scanner_heal_g14_multiset_evidence.py` | dev-tool | Assembles measured Scanner/Heal G14 same-window EC8+4 multi-set/multi-pool release descriptors from e2e case directories or an operator-collected proof | `.config/scanner-heal-required-tests.json`; `test_scanner_heal_g14_multiset_evidence.sh` | | `run_scanner_heal_g09_upgrade_evidence.sh` | dev-tool | Runs the G09 mixed-version and rollback upgrade E2E lanes against a pinned previous release and verifies the raw evidence artifacts | `docs/testing/ci-gates.md`; `.github/workflows/e2e-upgrade.yml`; `test_scanner_heal_g09_upgrade_evidence.sh` | | `run_scanner_heal_scheduler_pressure_evidence.py` | dev-tool | Assembles measured Scanner/Heal G10/P1/P3 scheduler-pressure release descriptors from a completed measured ABBA run, recovery-window proof, and profile artifacts | `docs/operations/scanner-benchmark-runbook.md`; `test_scanner_heal_scheduler_pressure_evidence.sh` | +| `run_scanner_heal_status_outcome_evidence.py` | dev-tool | Assembles measured Scanner/Heal G05/G06/R-D status-and-outcome release descriptors from status, compatibility, and disposition artifacts | `.config/scanner-heal-required-tests.json`; `test_scanner_heal_status_outcome_evidence.sh` | +| `run_scanner_heal_maintenance_evidence.py` | dev-tool | Assembles measured Scanner/Heal G11/G13 maintenance-producer release descriptors from operator-collected proof JSON | `.config/scanner-heal-required-tests.json`; `test_scanner_heal_maintenance_evidence.sh` | | `run_scanner_heal_w13_mrf_evidence.sh` | dev-tool | Runs the W13 durable MRF replay lanes and writes G07/G08/P4 bundle-ready evidence descriptors | `docs/testing/ci-gates.md`; `test_scanner_heal_w13_mrf_evidence.sh` | | `run_scanner_heal_w16_recovery_evidence.sh` | dev-tool | Runs the W16 recovery-intent and quota authority lanes and writes G04/G12 bundle-ready evidence descriptors | `docs/testing/ci-gates.md`; `test_scanner_heal_w16_recovery_evidence.sh` | | `test_scanner_validation_harness.sh` | dev-tool | Self-test for the scanner validation harness | — | +| `test_scanner_heal_checkpoint_crash_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal checkpoint/crash evidence assembler | — | | `test_scanner_heal_g14_multiset_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal G14 multi-set/multi-pool evidence assembler | — | | `test_scanner_heal_scheduler_pressure_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal scheduler-pressure evidence assembler | — | +| `test_scanner_heal_status_outcome_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal status-and-outcome evidence assembler | — | +| `test_scanner_heal_maintenance_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal G11/G13 maintenance evidence assembler | — | | `test_scanner_heal_g09_upgrade_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal G09 upgrade evidence runner | — | | `test_scanner_heal_w16_recovery_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal W16 recovery evidence runner | — | | `test_scanner_heal_w13_mrf_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal W13 MRF evidence runner | — | diff --git a/scripts/check_test_wiring.py b/scripts/check_test_wiring.py index 4ce267232..035ffbcaa 100755 --- a/scripts/check_test_wiring.py +++ b/scripts/check_test_wiring.py @@ -44,6 +44,15 @@ SCANNER_HEAL_RELEASE_REQUIRED_GATES = ( "P1", "P2", "P3", "P4", "R-E", "R-D", "R-L", ) SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS = { + "G05": ( + "per_object_outcome_oracle", + "terminal_retention_bounds", + ), + "G06": ( + "concurrent_status_evidence", + "legacy_client_compatibility", + "truncation_behavior", + ), "G07": ( "mrf_responsibility_oracle", "commit_boundary_crash_matrix", @@ -86,14 +95,20 @@ SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS = { "post_stop_convergence_measurement", "cold_segment_reuse_measurement", ), + "R-D": ( + "manager_disposition_evidence", + "event_disposition_evidence", + "ledger_disposition_evidence", + "grace_handling", + ), } SCANNER_HEAL_RELEASE_BUNDLE_REQUIRED_EVIDENCE_FIELDS = { "G01": ("root_authority_evidence", "quota_authority_evidence"), "G02": ("bounded_checkpoint_oracle", "independent_version_inventory"), "G03": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["G03"], "G04": ("cache_boundary_crash_evidence", "root_floor_intent_crash_evidence"), - "G05": ("per_object_outcome_oracle", "terminal_retention_bounds"), - "G06": ("concurrent_status_evidence", "legacy_client_compatibility", "truncation_behavior"), + "G05": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["G05"], + "G06": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["G06"], "G07": ("mrf_responsibility_oracle", "commit_boundary_crash_matrix"), "G08": ("mrf_capacity_evidence", "disk_full_matrix", "replica_loss_matrix"), "G09": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["G09"], @@ -107,7 +122,7 @@ SCANNER_HEAL_RELEASE_BUNDLE_REQUIRED_EVIDENCE_FIELDS = { "P3": ("two_hour_pressure_measurement", "heal_capacity_measurement", "recovery_window_measurement"), "P4": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["P4"], "R-E": ("fixed_budget_restart_evidence", "enumeration_evidence", "classification_evidence"), - "R-D": ("manager_disposition_evidence", "event_disposition_evidence", "ledger_disposition_evidence", "grace_handling"), + "R-D": SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["R-D"], "R-L": ("legacy_source_conflict_evidence", "migration_gap_evidence", "crash_safe_source_retirement_evidence"), } SCANNER_HEAL_RELEASE_MIXED_VERSION_ROLES = { @@ -157,6 +172,55 @@ SCANNER_HEAL_SEGMENT_ACTIVATION_PROOF_INPUTS = ( "generation_window", "producer_identities", ) +SCANNER_HEAL_REQUIRED_PRODUCER_IDENTITIES = ( + "put_object", + "delete_object", + "delete_marker", + "complete_multipart_upload", + "abort_multipart_upload", + "object_metadata", + "bucket_metadata", + "replication", + "tier_transition", + "tier_expiration", + "directory_object", +) +SCANNER_HEAL_REQUIRED_PRODUCER_FAMILIES = ( + "put", + "delete", + "delete_marker", + "multipart", + "replication", + "tier", + "directory_object", +) +SCANNER_HEAL_RELEASE_G11_REQUIRED_CASES = { + "maintenance_producer_matrix": ( + "object-mutation-producers", + "metadata-mutation-producers", + "replication-tier-producers", + "directory-object-producer", + ), + "complete_producer_inventory": SCANNER_HEAL_REQUIRED_PRODUCER_IDENTITIES, +} +SCANNER_HEAL_RELEASE_G13_REQUIRED_CASES = { + "quorum_minus_one_matrix": ( + "read-quorum-minus-one", + "write-quorum-minus-one", + "heal-quorum-minus-one", + "restart-quorum-minus-one", + ), + "unknown_disk_remount_matrix": ( + "unknown-disk-excluded-from-quorum", + "known-disk-remount-rejoins", + "stale-disk-remount-rejected", + ), + "object_lock_dry_run_grace_evidence": ( + "object-lock-delete-denied", + "dry-run-does-not-mutate", + "grace-window-retains-terminal-outcome", + ), +} SCANNER_HEAL_RELEASE_G07_REQUIRED_CASES = { "mrf_responsibility_oracle": ( "legacy-journal-replay", @@ -202,6 +266,58 @@ SCANNER_HEAL_RELEASE_CRASH_BOUNDARY_FIELDS = { "process-restart-replay", ), } +SCANNER_HEAL_RELEASE_G05_PER_OBJECT_OUTCOME_CASES = ( + "object-repaired", + "object-already-healthy", + "object-skipped-by-policy", + "object-failed-and-retained", +) +SCANNER_HEAL_RELEASE_G05_TERMINAL_RETENTION_CASES = ( + "finished-retained-until-window", + "failed-retained-until-window", + "canceled-retained-until-window", + "expired-terminal-pruned-after-window", +) +SCANNER_HEAL_RELEASE_G06_CONCURRENT_STATUS_CASES = ( + "status-during-admin-heal", + "status-during-background-heal", + "status-while-peer-down", + "status-after-peer-rejoin", +) +SCANNER_HEAL_RELEASE_G06_LEGACY_CLIENT_CASES = ( + "rustfs-admin-v3-background-heal-status", + "minio-admin-v3-background-heal-status", + "heal-client-token-empty-body", + "node-heal-status-v1-wire", +) +SCANNER_HEAL_RELEASE_G06_TRUNCATION_CASES = ( + "oversize-node-status-reject", + "truncated-node-status-reject", + "trailing-data-node-status-reject", +) +SCANNER_HEAL_RELEASE_RD_MANAGER_CASES = ( + "accepted", + "coalesced-duplicate", + "rejected-policy", + "terminal-retained", +) +SCANNER_HEAL_RELEASE_RD_EVENT_CASES = ( + "event-repaired", + "event-failed", + "event-skipped", + "event-grace-retained", +) +SCANNER_HEAL_RELEASE_RD_LEDGER_CASES = ( + "ledger-recorded", + "ledger-replayed", + "ledger-discharged", + "ledger-pruned-after-grace", +) +SCANNER_HEAL_RELEASE_RD_GRACE_CASES = ( + "grace-open-retains-disposition", + "grace-expired-prunes-terminal", + "restart-preserves-grace-clock", +) SCANNER_HEAL_RELEASE_SCOPED_ACK_CASES = { "durable_root_publication_proof": ( "root-cas-success", @@ -1563,8 +1679,49 @@ def release_bundle_json_artifact_mirrored_fields(gate: str, field: str) -> tuple fields: list[str] = [] if gate in ("G03", "G09", "R-L"): fields.extend(("versions", "mixed_version_role")) + if gate == "G02": + if field == "bounded_checkpoint_oracle": + fields.extend(( + "checkpoint_progress_bounded", + "raw_entry_budget", + "max_raw_entries_per_round", + "max_objects_processed_per_round", + "durable_checkpoint_committed", + "no_unbounded_tail", + )) + if field == "independent_version_inventory": + fields.extend(( + "independent_version_inventory_observed", + "objects_expected", + "objects_retained", + "versions_retained", + "bytes_retained", + )) if gate in ("G04", "G07", "R-E", "R-L"): fields.append("crash_points") + if gate == "R-E": + if field == "fixed_budget_restart_evidence": + fields.extend(( + "fixed_budget_restart_converged", + "restart_rounds", + "raw_entry_budget", + "no_unbudgeted_final_sweep", + )) + if field == "enumeration_evidence": + fields.extend(( + "raw_enumeration_observed", + "durable_raw_page_commit_observed", + "raw_page_index_complete", + "enumeration_frontier_retained", + )) + if field == "classification_evidence": + fields.extend(( + "classification_observed", + "objects_processed", + "objects_retained", + "versions_retained", + "bytes_retained", + )) if gate == "G03": fields.append("scoped_ack_cases") if field == "durable_root_publication_proof": @@ -1573,6 +1730,32 @@ def release_bundle_json_artifact_mirrored_fields(gate: str, field: str) -> tuple fields.append("whole_cycle_fallback_observed") if gate == "G04" and field == "root_floor_intent_crash_evidence": fields.extend(("durable_intent_cases", "persist_failure_blocks_acceptance")) + if gate == "G05": + if field == "per_object_outcome_oracle": + fields.extend(("per_object_outcome_cases", "outcome_counts", "status_matches_object_oracle")) + if field == "terminal_retention_bounds": + fields.extend(( + "terminal_retention_cases", + "terminal_retention_window_seconds", + "max_terminal_record_age_seconds", + "terminal_records_pruned_after_window", + )) + if gate == "G06": + if field == "concurrent_status_evidence": + fields.extend(( + "concurrent_status_cases", + "status_samples", + "all_status_responses_http_success", + "partial_status_reports_degraded", + )) + if field == "legacy_client_compatibility": + fields.extend(( + "legacy_client_cases", + "rustfs_and_minio_paths_compatible", + "empty_body_status_requests_accepted", + )) + if field == "truncation_behavior": + fields.extend(("truncation_cases", "truncated_payloads_rejected", "max_status_payload_bytes")) if gate == "G07": fields.append({ "mrf_responsibility_oracle": "mrf_responsibility_cases", @@ -1588,6 +1771,52 @@ def release_bundle_json_artifact_mirrored_fields(gate: str, field: str) -> tuple fields.append("mixed_version_cases") if field == "rollback_payload_evidence": fields.append("rollback_payload_replayed") + if gate == "G11": + fields.extend({ + "maintenance_producer_matrix": ( + "producer_identities", + "producer_families", + "matrix_cases", + "durable_identity_observed", + "generation_window_bound", + "restart_gap_absent", + "overflow_absent", + ), + "complete_producer_inventory": ( + "required_producer_identities", + "observed_producer_identities", + "required_producer_families", + "observed_producer_families", + "missing_producer_identities", + "unknown_producer_excluded", + ), + "segment_activation_preflight": ( + "production_activation", + "scanner_segment_reuse_activated", + "proof_inputs", + "fail_closed_checks", + ), + }[field]) + if gate == "G13": + fields.extend({ + "quorum_minus_one_matrix": ( + "quorum_cases", + "no_success_at_quorum_minus_one", + "exact_quorum_restored", + ), + "unknown_disk_remount_matrix": ( + "remount_cases", + "unknown_disks_excluded", + "remounted_disks_revalidated", + "stale_incarnation_rejected", + ), + "object_lock_dry_run_grace_evidence": ( + "grace_cases", + "object_lock_denials_preserved", + "dry_run_mutation_count", + "grace_outcomes_retained", + ), + }[field]) if (gate, field) in SCANNER_HEAL_RELEASE_MRF_DURABLE_REPLAY_FIELDS: fields.extend(("replayed_records", "responsibility_anchor_retained", "successor_snapshot_published")) if gate == "P4" and field == "retained_responsibility_evidence": @@ -1606,6 +1835,24 @@ def release_bundle_json_artifact_mirrored_fields(gate: str, field: str) -> tuple "pending_responsibilities_after_gc", "stale_journals_after_gc", )) + if gate == "R-D": + if field == "manager_disposition_evidence": + fields.extend(("manager_disposition_cases", "manager_dispositions_are_terminal")) + if field == "event_disposition_evidence": + fields.extend(("event_disposition_cases", "events_correlate_to_manager_dispositions")) + if field == "ledger_disposition_evidence": + fields.extend(( + "ledger_disposition_cases", + "ledger_correlates_to_events", + "ledger_replay_preserves_terminal_disposition", + )) + if field == "grace_handling": + fields.extend(( + "grace_cases", + "grace_window_seconds", + "grace_retention_observed", + "grace_expiry_pruned_terminal_records", + )) return tuple(dict.fromkeys(fields)) @@ -1706,6 +1953,38 @@ def release_bundle_number(value: object, name: str, minimum: int | float = 0) -> def validate_release_bundle_domain_evidence(gate: str, field: str, evidence: dict[str, object]) -> None: + if gate == "G02": + if field == "bounded_checkpoint_oracle": + release_bundle_bool_true(evidence.get("checkpoint_progress_bounded"), + f"{gate}.{field}.checkpoint_progress_bounded") + budget = evidence_integer(evidence.get("raw_entry_budget"), f"{gate}.{field}.raw_entry_budget", 1, 4096) + max_raw = evidence_integer(evidence.get("max_raw_entries_per_round"), + f"{gate}.{field}.max_raw_entries_per_round", 1, 4096) + max_objects = evidence_integer(evidence.get("max_objects_processed_per_round"), + f"{gate}.{field}.max_objects_processed_per_round", 1, 4096) + require(max_raw <= budget, f"{gate}.{field} raw entries exceed fixed budget") + require(max_objects <= budget, f"{gate}.{field} processed objects exceed fixed budget") + release_bundle_bool_true(evidence.get("durable_checkpoint_committed"), + f"{gate}.{field}.durable_checkpoint_committed") + release_bundle_bool_true(evidence.get("no_unbounded_tail"), f"{gate}.{field}.no_unbounded_tail") + if field == "independent_version_inventory": + release_bundle_bool_true(evidence.get("independent_version_inventory_observed"), + f"{gate}.{field}.independent_version_inventory_observed") + objects_expected = evidence_integer(evidence.get("objects_expected"), + f"{gate}.{field}.objects_expected", 1, 2**63 - 1) + objects_retained = evidence_integer(evidence.get("objects_retained"), + f"{gate}.{field}.objects_retained", 1, 2**63 - 1) + versions_retained = evidence_integer(evidence.get("versions_retained"), + f"{gate}.{field}.versions_retained", 1, 2**63 - 1) + bytes_retained = evidence_integer(evidence.get("bytes_retained"), + f"{gate}.{field}.bytes_retained", 1, 2**63 - 1) + require(objects_retained == objects_expected, + f"{gate}.{field} retained object inventory must match expected objects") + require(versions_retained == objects_expected, + f"{gate}.{field} retained version inventory must match expected objects") + require(bytes_retained == objects_expected, + f"{gate}.{field} retained byte inventory must match expected objects") + if gate == "G03": release_bundle_exact_strings( evidence.get("scoped_ack_cases"), @@ -1728,6 +2007,108 @@ def validate_release_bundle_domain_evidence(gate: str, field: str, evidence: dic release_bundle_bool_true(evidence.get("persist_failure_blocks_acceptance"), f"{gate}.{field}.persist_failure_blocks_acceptance") + if gate == "G05": + if field == "per_object_outcome_oracle": + release_bundle_exact_strings( + evidence.get("per_object_outcome_cases"), + SCANNER_HEAL_RELEASE_G05_PER_OBJECT_OUTCOME_CASES, + f"{gate}.{field}.per_object_outcome_cases", + ) + outcomes = evidence.get("outcome_counts") + require(isinstance(outcomes, dict), f"{gate}.{field} missing outcome counts") + for outcome in ("repaired", "healthy", "skipped", "failed"): + evidence_integer(outcomes.get(outcome), f"{gate}.{field}.outcome_counts.{outcome}", 1, 2**63 - 1) + release_bundle_bool_true(evidence.get("status_matches_object_oracle"), + f"{gate}.{field}.status_matches_object_oracle") + if field == "terminal_retention_bounds": + release_bundle_exact_strings( + evidence.get("terminal_retention_cases"), + SCANNER_HEAL_RELEASE_G05_TERMINAL_RETENTION_CASES, + f"{gate}.{field}.terminal_retention_cases", + ) + retention_window = evidence_integer( + evidence.get("terminal_retention_window_seconds"), + f"{gate}.{field}.terminal_retention_window_seconds", + 1, + 86400, + ) + max_age = evidence_integer( + evidence.get("max_terminal_record_age_seconds"), + f"{gate}.{field}.max_terminal_record_age_seconds", + 0, + 86400, + ) + require(max_age <= retention_window, f"{gate}.{field}.max_terminal_record_age_seconds exceeds retention window") + evidence_integer(evidence.get("terminal_records_pruned_after_window"), + f"{gate}.{field}.terminal_records_pruned_after_window", 1, 2**63 - 1) + + if gate == "G06": + if field == "concurrent_status_evidence": + release_bundle_exact_strings( + evidence.get("concurrent_status_cases"), + SCANNER_HEAL_RELEASE_G06_CONCURRENT_STATUS_CASES, + f"{gate}.{field}.concurrent_status_cases", + ) + evidence_integer(evidence.get("status_samples"), f"{gate}.{field}.status_samples", 2, 2**63 - 1) + release_bundle_bool_true(evidence.get("all_status_responses_http_success"), + f"{gate}.{field}.all_status_responses_http_success") + release_bundle_bool_true(evidence.get("partial_status_reports_degraded"), + f"{gate}.{field}.partial_status_reports_degraded") + if field == "legacy_client_compatibility": + release_bundle_exact_strings( + evidence.get("legacy_client_cases"), + SCANNER_HEAL_RELEASE_G06_LEGACY_CLIENT_CASES, + f"{gate}.{field}.legacy_client_cases", + ) + release_bundle_bool_true(evidence.get("rustfs_and_minio_paths_compatible"), + f"{gate}.{field}.rustfs_and_minio_paths_compatible") + release_bundle_bool_true(evidence.get("empty_body_status_requests_accepted"), + f"{gate}.{field}.empty_body_status_requests_accepted") + if field == "truncation_behavior": + release_bundle_exact_strings( + evidence.get("truncation_cases"), + SCANNER_HEAL_RELEASE_G06_TRUNCATION_CASES, + f"{gate}.{field}.truncation_cases", + ) + release_bundle_bool_true(evidence.get("truncated_payloads_rejected"), + f"{gate}.{field}.truncated_payloads_rejected") + evidence_integer(evidence.get("max_status_payload_bytes"), f"{gate}.{field}.max_status_payload_bytes", 1, 2**20) + + if gate == "R-E": + if field == "fixed_budget_restart_evidence": + release_bundle_bool_true(evidence.get("fixed_budget_restart_converged"), + f"{gate}.{field}.fixed_budget_restart_converged") + evidence_integer(evidence.get("restart_rounds"), f"{gate}.{field}.restart_rounds", 2, 64) + evidence_integer(evidence.get("raw_entry_budget"), f"{gate}.{field}.raw_entry_budget", 1, 4096) + release_bundle_bool_true(evidence.get("no_unbudgeted_final_sweep"), + f"{gate}.{field}.no_unbudgeted_final_sweep") + if field == "enumeration_evidence": + release_bundle_bool_true(evidence.get("raw_enumeration_observed"), + f"{gate}.{field}.raw_enumeration_observed") + release_bundle_bool_true(evidence.get("durable_raw_page_commit_observed"), + f"{gate}.{field}.durable_raw_page_commit_observed") + release_bundle_bool_true(evidence.get("raw_page_index_complete"), + f"{gate}.{field}.raw_page_index_complete") + release_bundle_bool_true(evidence.get("enumeration_frontier_retained"), + f"{gate}.{field}.enumeration_frontier_retained") + if field == "classification_evidence": + release_bundle_bool_true(evidence.get("classification_observed"), + f"{gate}.{field}.classification_observed") + objects_processed = evidence_integer(evidence.get("objects_processed"), + f"{gate}.{field}.objects_processed", 1, 2**63 - 1) + objects_retained = evidence_integer(evidence.get("objects_retained"), + f"{gate}.{field}.objects_retained", 1, 2**63 - 1) + versions_retained = evidence_integer(evidence.get("versions_retained"), + f"{gate}.{field}.versions_retained", 1, 2**63 - 1) + bytes_retained = evidence_integer(evidence.get("bytes_retained"), + f"{gate}.{field}.bytes_retained", 1, 2**63 - 1) + require(objects_processed <= objects_retained, + f"{gate}.{field} processed objects exceed retained objects") + require(versions_retained == objects_retained, + f"{gate}.{field} version inventory must match retained objects") + require(bytes_retained == objects_retained, + f"{gate}.{field} byte inventory must match retained objects") + if gate == "G09": release_bundle_exact_strings( evidence.get("mixed_version_cases"), @@ -1757,6 +2138,89 @@ def validate_release_bundle_domain_evidence(gate: str, field: str, evidence: dic evidence_integer(metrics.get("foreground_pressure_high_samples"), f"{gate}.{field}.foreground_pressure_high_samples", 1, 2**63 - 1) + if gate == "G11": + if field == "maintenance_producer_matrix": + release_bundle_exact_strings( + evidence.get("producer_identities"), + SCANNER_HEAL_REQUIRED_PRODUCER_IDENTITIES, + f"{gate}.{field}.producer_identities", + ) + release_bundle_exact_strings( + evidence.get("producer_families"), + SCANNER_HEAL_REQUIRED_PRODUCER_FAMILIES, + f"{gate}.{field}.producer_families", + ) + release_bundle_exact_strings( + evidence.get("matrix_cases"), + SCANNER_HEAL_RELEASE_G11_REQUIRED_CASES[field], + f"{gate}.{field}.matrix_cases", + ) + release_bundle_bool_true(evidence.get("durable_identity_observed"), + f"{gate}.{field}.durable_identity_observed") + release_bundle_bool_true(evidence.get("generation_window_bound"), + f"{gate}.{field}.generation_window_bound") + release_bundle_bool_true(evidence.get("restart_gap_absent"), f"{gate}.{field}.restart_gap_absent") + release_bundle_bool_true(evidence.get("overflow_absent"), f"{gate}.{field}.overflow_absent") + if field == "complete_producer_inventory": + release_bundle_exact_strings( + evidence.get("required_producer_identities"), + SCANNER_HEAL_REQUIRED_PRODUCER_IDENTITIES, + f"{gate}.{field}.required_producer_identities", + ) + release_bundle_exact_strings( + evidence.get("observed_producer_identities"), + SCANNER_HEAL_REQUIRED_PRODUCER_IDENTITIES, + f"{gate}.{field}.observed_producer_identities", + ) + release_bundle_exact_strings( + evidence.get("required_producer_families"), + SCANNER_HEAL_REQUIRED_PRODUCER_FAMILIES, + f"{gate}.{field}.required_producer_families", + ) + release_bundle_exact_strings( + evidence.get("observed_producer_families"), + SCANNER_HEAL_REQUIRED_PRODUCER_FAMILIES, + f"{gate}.{field}.observed_producer_families", + ) + require(evidence.get("missing_producer_identities") == [], + f"{gate}.{field} requires zero missing producer identities") + release_bundle_bool_true(evidence.get("unknown_producer_excluded"), + f"{gate}.{field}.unknown_producer_excluded") + + if gate == "G13": + if field == "quorum_minus_one_matrix": + release_bundle_exact_strings( + evidence.get("quorum_cases"), + SCANNER_HEAL_RELEASE_G13_REQUIRED_CASES[field], + f"{gate}.{field}.quorum_cases", + ) + release_bundle_bool_true(evidence.get("no_success_at_quorum_minus_one"), + f"{gate}.{field}.no_success_at_quorum_minus_one") + release_bundle_bool_true(evidence.get("exact_quorum_restored"), f"{gate}.{field}.exact_quorum_restored") + if field == "unknown_disk_remount_matrix": + release_bundle_exact_strings( + evidence.get("remount_cases"), + SCANNER_HEAL_RELEASE_G13_REQUIRED_CASES[field], + f"{gate}.{field}.remount_cases", + ) + release_bundle_bool_true(evidence.get("unknown_disks_excluded"), + f"{gate}.{field}.unknown_disks_excluded") + release_bundle_bool_true(evidence.get("remounted_disks_revalidated"), + f"{gate}.{field}.remounted_disks_revalidated") + release_bundle_bool_true(evidence.get("stale_incarnation_rejected"), + f"{gate}.{field}.stale_incarnation_rejected") + if field == "object_lock_dry_run_grace_evidence": + release_bundle_exact_strings( + evidence.get("grace_cases"), + SCANNER_HEAL_RELEASE_G13_REQUIRED_CASES[field], + f"{gate}.{field}.grace_cases", + ) + release_bundle_bool_true(evidence.get("object_lock_denials_preserved"), + f"{gate}.{field}.object_lock_denials_preserved") + evidence_integer(evidence.get("dry_run_mutation_count"), f"{gate}.{field}.dry_run_mutation_count", 0, 0) + release_bundle_bool_true(evidence.get("grace_outcomes_retained"), + f"{gate}.{field}.grace_outcomes_retained") + if gate == "P1": if field == "cold_walk_share_measurement": share = release_bundle_number(evidence.get("cold_walk_share"), f"{gate}.{field}.cold_walk_share") @@ -1811,6 +2275,45 @@ def validate_release_bundle_domain_evidence(gate: str, field: str, evidence: dic f"{gate}.{field}.verified_proof_discharge_observed", ) + if gate == "R-D": + if field == "manager_disposition_evidence": + release_bundle_exact_strings( + evidence.get("manager_disposition_cases"), + SCANNER_HEAL_RELEASE_RD_MANAGER_CASES, + f"{gate}.{field}.manager_disposition_cases", + ) + release_bundle_bool_true(evidence.get("manager_dispositions_are_terminal"), + f"{gate}.{field}.manager_dispositions_are_terminal") + if field == "event_disposition_evidence": + release_bundle_exact_strings( + evidence.get("event_disposition_cases"), + SCANNER_HEAL_RELEASE_RD_EVENT_CASES, + f"{gate}.{field}.event_disposition_cases", + ) + release_bundle_bool_true(evidence.get("events_correlate_to_manager_dispositions"), + f"{gate}.{field}.events_correlate_to_manager_dispositions") + if field == "ledger_disposition_evidence": + release_bundle_exact_strings( + evidence.get("ledger_disposition_cases"), + SCANNER_HEAL_RELEASE_RD_LEDGER_CASES, + f"{gate}.{field}.ledger_disposition_cases", + ) + release_bundle_bool_true(evidence.get("ledger_correlates_to_events"), + f"{gate}.{field}.ledger_correlates_to_events") + release_bundle_bool_true(evidence.get("ledger_replay_preserves_terminal_disposition"), + f"{gate}.{field}.ledger_replay_preserves_terminal_disposition") + if field == "grace_handling": + release_bundle_exact_strings( + evidence.get("grace_cases"), + SCANNER_HEAL_RELEASE_RD_GRACE_CASES, + f"{gate}.{field}.grace_cases", + ) + evidence_integer(evidence.get("grace_window_seconds"), f"{gate}.{field}.grace_window_seconds", 1, 86400) + release_bundle_bool_true(evidence.get("grace_retention_observed"), + f"{gate}.{field}.grace_retention_observed") + release_bundle_bool_true(evidence.get("grace_expiry_pruned_terminal_records"), + f"{gate}.{field}.grace_expiry_pruned_terminal_records") + def validate_release_bundle_artifact(bundle_path: Path, source_revision: str, gate: str, field: str, evidence: dict[str, object]) -> str: @@ -1858,6 +2361,8 @@ def validate_release_bundle_artifact(bundle_path: Path, source_revision: str, ga evidence_integer(evidence.get("lock_hold_p95_ms"), f"{gate}.{field}.lock_hold_p95_ms", 0, 2**31 - 1) evidence_integer(evidence.get("foreground_latency_p95_ms"), f"{gate}.{field}.foreground_latency_p95_ms", 1, 2**31 - 1) + if gate in ("G11", "G13"): + validate_release_bundle_domain_evidence(gate, field, evidence) if gate == "P1" and field == "foreground_latency_throughput_measurement": evidence_integer(evidence.get("foreground_latency_p95_ms"), f"{gate}.{field}.foreground_latency_p95_ms", 1, 2**31 - 1) @@ -1932,6 +2437,25 @@ def validate_release_bundle_artifact(bundle_path: Path, source_revision: str, ga f"{gate}.{field} requires full-walk oracle equivalence") require(evidence.get("published_root_equivalent") is True, f"{gate}.{field} requires published-root equivalence") + if field == "post_stop_convergence_measurement": + require(evidence.get("writes_stopped") is True, + f"{gate}.{field} requires writes-stopped evidence") + require(evidence.get("last_mutation_observed") is True, + f"{gate}.{field} requires last-mutation observation") + require(evidence.get("first_complete_publication") is True, + f"{gate}.{field} requires first complete publication evidence") + samples = evidence_integer(evidence.get("post_stop_samples"), f"{gate}.{field}.post_stop_samples", 1, 2**31 - 1) + multiple = release_bundle_number(evidence.get("post_stop_work_multiple"), + f"{gate}.{field}.post_stop_work_multiple", 0) + limit = release_bundle_number(evidence.get("post_stop_work_multiple_limit"), + f"{gate}.{field}.post_stop_work_multiple_limit", 1) + require(multiple <= limit, f"{gate}.{field} exceeds post-stop work multiple limit") + raw_multiples = evidence.get("post_stop_work_multiples") + require(isinstance(raw_multiples, list) and len(raw_multiples) == samples, + f"{gate}.{field} requires measured post-stop work multiple samples") + observed = [release_bundle_number(value, f"{gate}.{field}.post_stop_work_multiples", 0) + for value in raw_multiples] + require(max(observed) == multiple, f"{gate}.{field} worst post-stop multiple mismatch") if gate == "G07": case_field = { "mrf_responsibility_oracle": "mrf_responsibility_cases", @@ -2343,6 +2867,45 @@ def write_scanner_heal_release_bundle_fixture(root: Path, directory: Path) -> Pa evidence.update({"completed_heal_objects": 1, "duplicate_task_count": 0}) if gate == "P3" and field == "recovery_window_measurement": evidence.update({"pressure_recovery_window_seconds": 5, "lock_hold_p95_ms": 0}) + if gate == "G02" and field == "bounded_checkpoint_oracle": + evidence.update({ + "checkpoint_progress_bounded": True, + "raw_entry_budget": 8, + "max_raw_entries_per_round": 8, + "max_objects_processed_per_round": 8, + "durable_checkpoint_committed": True, + "no_unbounded_tail": True, + }) + if gate == "G02" and field == "independent_version_inventory": + evidence.update({ + "independent_version_inventory_observed": True, + "objects_expected": 16, + "objects_retained": 16, + "versions_retained": 16, + "bytes_retained": 16, + }) + if gate == "R-E" and field == "fixed_budget_restart_evidence": + evidence.update({ + "fixed_budget_restart_converged": True, + "restart_rounds": 3, + "raw_entry_budget": 8, + "no_unbudgeted_final_sweep": True, + }) + if gate == "R-E" and field == "enumeration_evidence": + evidence.update({ + "raw_enumeration_observed": True, + "durable_raw_page_commit_observed": True, + "raw_page_index_complete": True, + "enumeration_frontier_retained": True, + }) + if gate == "R-E" and field == "classification_evidence": + evidence.update({ + "classification_observed": True, + "objects_processed": 16, + "objects_retained": 16, + "versions_retained": 16, + "bytes_retained": 16, + }) if gate in ("G03", "G09", "R-L"): evidence["versions"] = [baseline_revision, source_revision] evidence["mixed_version_role"] = SCANNER_HEAL_RELEASE_MIXED_VERSION_ROLES[(gate, field)] @@ -2718,6 +3281,23 @@ class SelfTests(unittest.TestCase): evidence.update({"completed_heal_objects": 1, "duplicate_task_count": 0}) if gate == "P3" and field == "recovery_window_measurement": evidence.update({"pressure_recovery_window_seconds": 5, "lock_hold_p95_ms": 0}) + if gate == "G02" and field == "bounded_checkpoint_oracle": + evidence.update({ + "checkpoint_progress_bounded": True, + "raw_entry_budget": 8, + "max_raw_entries_per_round": 8, + "max_objects_processed_per_round": 8, + "durable_checkpoint_committed": True, + "no_unbounded_tail": True, + }) + if gate == "G02" and field == "independent_version_inventory": + evidence.update({ + "independent_version_inventory_observed": True, + "objects_expected": 16, + "objects_retained": 16, + "versions_retained": 16, + "bytes_retained": 16, + }) if gate in ("G03", "G09", "R-L"): evidence["versions"] = ["a" * 40, source_revision] evidence["mixed_version_role"] = SCANNER_HEAL_RELEASE_MIXED_VERSION_ROLES[(gate, field)] @@ -2733,6 +3313,50 @@ class SelfTests(unittest.TestCase): if gate == "G04" and field == "root_floor_intent_crash_evidence": evidence["durable_intent_cases"] = list(SCANNER_HEAL_RELEASE_CRASH_BOUNDARY_FIELDS[(gate, field)]) evidence["persist_failure_blocks_acceptance"] = True + if gate == "G05" and field == "per_object_outcome_oracle": + evidence["per_object_outcome_cases"] = list(SCANNER_HEAL_RELEASE_G05_PER_OBJECT_OUTCOME_CASES) + evidence["outcome_counts"] = {"repaired": 4, "healthy": 3, "skipped": 2, "failed": 1} + evidence["status_matches_object_oracle"] = True + if gate == "G05" and field == "terminal_retention_bounds": + evidence["terminal_retention_cases"] = list(SCANNER_HEAL_RELEASE_G05_TERMINAL_RETENTION_CASES) + evidence["terminal_retention_window_seconds"] = 3600 + evidence["max_terminal_record_age_seconds"] = 3599 + evidence["terminal_records_pruned_after_window"] = 2 + if gate == "G06" and field == "concurrent_status_evidence": + evidence["concurrent_status_cases"] = list(SCANNER_HEAL_RELEASE_G06_CONCURRENT_STATUS_CASES) + evidence["status_samples"] = 4 + evidence["all_status_responses_http_success"] = True + evidence["partial_status_reports_degraded"] = True + if gate == "G06" and field == "legacy_client_compatibility": + evidence["legacy_client_cases"] = list(SCANNER_HEAL_RELEASE_G06_LEGACY_CLIENT_CASES) + evidence["rustfs_and_minio_paths_compatible"] = True + evidence["empty_body_status_requests_accepted"] = True + if gate == "G06" and field == "truncation_behavior": + evidence["truncation_cases"] = list(SCANNER_HEAL_RELEASE_G06_TRUNCATION_CASES) + evidence["truncated_payloads_rejected"] = True + evidence["max_status_payload_bytes"] = 4096 + if gate == "R-E" and field == "fixed_budget_restart_evidence": + evidence.update({ + "fixed_budget_restart_converged": True, + "restart_rounds": 3, + "raw_entry_budget": 8, + "no_unbudgeted_final_sweep": True, + }) + if gate == "R-E" and field == "enumeration_evidence": + evidence.update({ + "raw_enumeration_observed": True, + "durable_raw_page_commit_observed": True, + "raw_page_index_complete": True, + "enumeration_frontier_retained": True, + }) + if gate == "R-E" and field == "classification_evidence": + evidence.update({ + "classification_observed": True, + "objects_processed": 16, + "objects_retained": 16, + "versions_retained": 16, + "bytes_retained": 16, + }) if gate == "G09": evidence["mixed_version_cases"] = list(SCANNER_HEAL_RELEASE_MIXED_VERSION_CASES[field]) if field == "rollback_payload_evidence": @@ -2773,6 +3397,35 @@ class SelfTests(unittest.TestCase): "foreground_pressure_samples": 120, "foreground_pressure_high_samples": 12, } + if gate == "G11" and field == "maintenance_producer_matrix": + evidence["producer_identities"] = list(SCANNER_HEAL_REQUIRED_PRODUCER_IDENTITIES) + evidence["producer_families"] = list(SCANNER_HEAL_REQUIRED_PRODUCER_FAMILIES) + evidence["matrix_cases"] = list(SCANNER_HEAL_RELEASE_G11_REQUIRED_CASES[field]) + evidence["durable_identity_observed"] = True + evidence["generation_window_bound"] = True + evidence["restart_gap_absent"] = True + evidence["overflow_absent"] = True + if gate == "G11" and field == "complete_producer_inventory": + evidence["required_producer_identities"] = list(SCANNER_HEAL_REQUIRED_PRODUCER_IDENTITIES) + evidence["observed_producer_identities"] = list(SCANNER_HEAL_REQUIRED_PRODUCER_IDENTITIES) + evidence["required_producer_families"] = list(SCANNER_HEAL_REQUIRED_PRODUCER_FAMILIES) + evidence["observed_producer_families"] = list(SCANNER_HEAL_REQUIRED_PRODUCER_FAMILIES) + evidence["missing_producer_identities"] = [] + evidence["unknown_producer_excluded"] = True + if gate == "G13" and field == "quorum_minus_one_matrix": + evidence["quorum_cases"] = list(SCANNER_HEAL_RELEASE_G13_REQUIRED_CASES[field]) + evidence["no_success_at_quorum_minus_one"] = True + evidence["exact_quorum_restored"] = True + if gate == "G13" and field == "unknown_disk_remount_matrix": + evidence["remount_cases"] = list(SCANNER_HEAL_RELEASE_G13_REQUIRED_CASES[field]) + evidence["unknown_disks_excluded"] = True + evidence["remounted_disks_revalidated"] = True + evidence["stale_incarnation_rejected"] = True + if gate == "G13" and field == "object_lock_dry_run_grace_evidence": + evidence["grace_cases"] = list(SCANNER_HEAL_RELEASE_G13_REQUIRED_CASES[field]) + evidence["object_lock_denials_preserved"] = True + evidence["dry_run_mutation_count"] = 0 + evidence["grace_outcomes_retained"] = True if gate == "G14" and field == "ec8_4_evidence": evidence["topology"] = {"erasure": "EC8+4", "nodes": 3, "drives_per_node": 4} if gate == "G14" and field == "same_window_field_evidence": @@ -2801,6 +3454,14 @@ class SelfTests(unittest.TestCase): evidence["cold_walked_segments"] = 0 evidence["full_walk_oracle_equivalent"] = True evidence["published_root_equivalent"] = True + if field == "post_stop_convergence_measurement": + evidence["writes_stopped"] = True + evidence["last_mutation_observed"] = True + evidence["first_complete_publication"] = True + evidence["post_stop_samples"] = 2 + evidence["post_stop_work_multiple"] = 1.1 + evidence["post_stop_work_multiple_limit"] = 1.2 + evidence["post_stop_work_multiples"] = [1.0, 1.1] if gate == "P4" and field == "retained_responsibility_evidence": evidence["duration_seconds"] = 7200 evidence["finished_at"] = (started + timedelta(seconds=7200)).isoformat().replace("+00:00", "Z") @@ -2810,6 +3471,21 @@ class SelfTests(unittest.TestCase): evidence["retention_window_seconds"] = 7200 evidence["idle_cleanup_observed"] = True evidence["verified_proof_discharge_observed"] = True + if gate == "R-D" and field == "manager_disposition_evidence": + evidence["manager_disposition_cases"] = list(SCANNER_HEAL_RELEASE_RD_MANAGER_CASES) + evidence["manager_dispositions_are_terminal"] = True + if gate == "R-D" and field == "event_disposition_evidence": + evidence["event_disposition_cases"] = list(SCANNER_HEAL_RELEASE_RD_EVENT_CASES) + evidence["events_correlate_to_manager_dispositions"] = True + if gate == "R-D" and field == "ledger_disposition_evidence": + evidence["ledger_disposition_cases"] = list(SCANNER_HEAL_RELEASE_RD_LEDGER_CASES) + evidence["ledger_correlates_to_events"] = True + evidence["ledger_replay_preserves_terminal_disposition"] = True + if gate == "R-D" and field == "grace_handling": + evidence["grace_cases"] = list(SCANNER_HEAL_RELEASE_RD_GRACE_CASES) + evidence["grace_window_seconds"] = 300 + evidence["grace_retention_observed"] = True + evidence["grace_expiry_pruned_terminal_records"] = True if field == "profile_evidence": evidence["resolved_samples"] = 1 evidence["allocation_bytes"] = 1024 @@ -3054,6 +3730,36 @@ class SelfTests(unittest.TestCase): ), "measurement window mismatch", ), + ("g05-outcome-cases", "G05", "per_object_outcome_oracle", lambda item: item["per_object_outcome_cases"].remove("object-failed-and-retained"), "missing cases"), + ("g05-outcome-counts", "G05", "per_object_outcome_oracle", lambda item: item["outcome_counts"].pop("failed"), "outcome_counts.failed"), + ( + "g05-retention-window", + "G05", + "terminal_retention_bounds", + lambda item: item.update({"max_terminal_record_age_seconds": item["terminal_retention_window_seconds"] + 1}), + "exceeds retention window", + ), + ( + "g06-concurrent-status", + "G06", + "concurrent_status_evidence", + lambda item: item["concurrent_status_cases"].remove("status-while-peer-down"), + "missing cases", + ), + ( + "g06-legacy-client", + "G06", + "legacy_client_compatibility", + lambda item: item.update({"empty_body_status_requests_accepted": False}), + "empty_body_status_requests_accepted", + ), + ( + "g06-truncation", + "G06", + "truncation_behavior", + lambda item: item["truncation_cases"].remove("truncated-node-status-reject"), + "missing cases", + ), ("versions", "G09", "mixed_version_reader_evidence", lambda item: item.update({"versions": [1, 2]}), "mixed-version"), ("stale-versions", "G09", "mixed_version_writer_evidence", lambda item: item.update({"versions": ["a" * 40, "c" * 40]}), "tested source revision"), ("g08-capacity-cases", "G08", "mrf_capacity_evidence", lambda item: item.update({"capacity_cases": ["queue-count-limit"]}), "missing cases"), @@ -3070,6 +3776,48 @@ class SelfTests(unittest.TestCase): ("mrf-records", "G07", "mrf_responsibility_oracle", lambda item: item.pop("replayed_records"), "replayed_records"), ("mrf-anchor", "G07", "commit_boundary_crash_matrix", lambda item: item.update({"responsibility_anchor_retained": False}), "retained MRF responsibility anchors"), ("mrf-successor", "P4", "retained_responsibility_evidence", lambda item: item.pop("successor_snapshot_published"), "successor snapshot"), + ( + "g02-raw-budget", + "G02", + "bounded_checkpoint_oracle", + lambda item: item.update({"max_raw_entries_per_round": item["raw_entry_budget"] + 1}), + "raw entries exceed fixed budget", + ), + ( + "g02-durable-checkpoint", + "G02", + "bounded_checkpoint_oracle", + lambda item: item.update({"durable_checkpoint_committed": False}), + "durable_checkpoint_committed", + ), + ( + "g02-version-inventory", + "G02", + "independent_version_inventory", + lambda item: item.update({"versions_retained": item["objects_retained"] - 1}), + "version inventory", + ), + ( + "re-converged", + "R-E", + "fixed_budget_restart_evidence", + lambda item: item.update({"fixed_budget_restart_converged": False}), + "fixed_budget_restart_converged", + ), + ( + "re-enumeration-commit", + "R-E", + "enumeration_evidence", + lambda item: item.update({"durable_raw_page_commit_observed": False}), + "durable_raw_page_commit_observed", + ), + ( + "re-classification-inventory", + "R-E", + "classification_evidence", + lambda item: item.update({"bytes_retained": item["objects_retained"] - 1}), + "byte inventory", + ), ( "mrf-cleanup-gc-cases", "P4", @@ -3114,6 +3862,48 @@ class SelfTests(unittest.TestCase): lambda item: item["proof_inputs"].remove("process_epoch"), "missing G11.segment_activation_preflight.proof_inputs", ), + ( + "producer-matrix-missing", + "G11", + "maintenance_producer_matrix", + lambda item: item["producer_identities"].remove("tier_expiration"), + "producer_identities missing cases", + ), + ( + "producer-inventory-missing", + "G11", + "complete_producer_inventory", + lambda item: item.update({"missing_producer_identities": ["tier_expiration"]}), + "zero missing producer identities", + ), + ( + "producer-inventory-unknown", + "G11", + "complete_producer_inventory", + lambda item: item.update({"unknown_producer_excluded": False}), + "unknown_producer_excluded", + ), + ( + "quorum-minus-one-success", + "G13", + "quorum_minus_one_matrix", + lambda item: item.update({"no_success_at_quorum_minus_one": False}), + "no_success_at_quorum_minus_one", + ), + ( + "remount-revalidation", + "G13", + "unknown_disk_remount_matrix", + lambda item: item.update({"remounted_disks_revalidated": False}), + "remounted_disks_revalidated", + ), + ( + "dry-run-mutates", + "G13", + "object_lock_dry_run_grace_evidence", + lambda item: item.update({"dry_run_mutation_count": 1}), + "dry_run_mutation_count", + ), ( "distributed-invalidation", "G14", @@ -3128,6 +3918,20 @@ class SelfTests(unittest.TestCase): lambda item: item.update({"all_peers_bound_to_generation_window": False}), "peer generation-window binding", ), + ( + "post-stop-work-multiple", + "P2", + "post_stop_convergence_measurement", + lambda item: item.update({"post_stop_work_multiple": 1.3}), + "post-stop work multiple limit", + ), + ( + "post-stop-publication", + "P2", + "post_stop_convergence_measurement", + lambda item: item.update({"first_complete_publication": False}), + "first complete publication", + ), ( "cold-segment-walk", "P2", @@ -3173,6 +3977,34 @@ class SelfTests(unittest.TestCase): ("scheduler-duplicates", "G10", "scheduler_bound_evidence", lambda item: item.pop("duplicate_task_count"), "duplicate_task_count"), ("p1-throughput", "P1", "foreground_latency_throughput_measurement", lambda item: item.pop("throughput_ops_per_second"), "throughput_ops_per_second"), ("p3-fixed-load", "P3", "two_hour_pressure_measurement", lambda item: item.update({"fixed_offered_load": False}), "fixed offered load"), + ( + "rd-manager", + "R-D", + "manager_disposition_evidence", + lambda item: item["manager_disposition_cases"].remove("rejected-policy"), + "missing cases", + ), + ( + "rd-event", + "R-D", + "event_disposition_evidence", + lambda item: item.update({"events_correlate_to_manager_dispositions": False}), + "events_correlate_to_manager_dispositions", + ), + ( + "rd-ledger", + "R-D", + "ledger_disposition_evidence", + lambda item: item.update({"ledger_replay_preserves_terminal_disposition": False}), + "ledger_replay_preserves_terminal_disposition", + ), + ( + "rd-grace", + "R-D", + "grace_handling", + lambda item: item["grace_cases"].remove("restart-preserves-grace-clock"), + "missing cases", + ), ): with self.subTest(fault=fault), tempfile.TemporaryDirectory() as tmp: root, bundle = self.scanner_heal_release_bundle_fixture(Path(tmp)) @@ -3260,6 +4092,18 @@ class SelfTests(unittest.TestCase): ("G08", "disk_full_matrix"), "JSON artifact is fixture", ), + ( + "g05-outcome-mirror", + lambda payload: payload["per_object_outcome_cases"].remove("object-skipped-by-policy"), + ("G05", "per_object_outcome_oracle"), + "per_object_outcome_cases missing cases", + ), + ( + "g06-truncation-mirror", + lambda payload: payload["truncation_cases"].remove("trailing-data-node-status-reject"), + ("G06", "truncation_behavior"), + "truncation_cases missing cases", + ), ( "mrf-artifact-kind", lambda payload: payload.update({"artifact_kind": "generic-json"}), @@ -3278,6 +4122,12 @@ class SelfTests(unittest.TestCase): ("P4", "mrf_cleanup_gc_soak_evidence"), "JSON artifact requires zero pending responsibilities", ), + ( + "rd-grace-mirror", + lambda payload: payload["grace_cases"].remove("grace-expired-prunes-terminal"), + ("R-D", "grace_handling"), + "grace_cases missing cases", + ), ): with self.subTest(fault=fault), tempfile.TemporaryDirectory() as tmp: root, bundle = self.scanner_heal_release_bundle_fixture(Path(tmp)) @@ -3516,6 +4366,8 @@ class SelfTests(unittest.TestCase): self.assertIn("scheduler-pressure", status["pending_lanes"]) requirements, _, _ = scanner_heal_release_requirements(read_json(root / ".config/scanner-heal-required-tests.json")) self.assertIn("durable_root_publication_proof", requirements["G03"]["evidence_fields"]) + self.assertIn("per_object_outcome_oracle", requirements["G05"]["evidence_fields"]) + self.assertIn("truncation_behavior", requirements["G06"]["evidence_fields"]) self.assertIn("disk_full_matrix", requirements["G08"]["evidence_fields"]) self.assertIn("mixed_version_writer_evidence", requirements["G09"]["evidence_fields"]) self.assertIn("segment_activation_preflight", requirements["G11"]["evidence_fields"]) @@ -3525,12 +4377,13 @@ class SelfTests(unittest.TestCase): tuple(requirements["P4"]["evidence_fields"]), SCANNER_HEAL_RELEASE_REQUIRED_EVIDENCE_FIELDS["P4"], ) + self.assertIn("grace_handling", requirements["R-D"]["evidence_fields"]) def test_scanner_heal_required_evidence_fields_cannot_be_removed(self) -> None: with tempfile.TemporaryDirectory() as tmp: root, run_dir = self.scanner_heal_fixture(Path(tmp)) registry = read_json(root / ".config/scanner-heal-required-tests.json") - for gate in ("G03", "G07", "G08", "G09", "G11", "G14", "P2", "P4"): + for gate in ("G03", "G05", "G06", "G07", "G08", "G09", "G11", "G14", "P2", "P4", "R-D"): for requirement in registry["release_requirements"]: if requirement["gate"] == gate: requirement["evidence_fields"] = [] diff --git a/scripts/run_scanner_heal_checkpoint_crash_evidence.py b/scripts/run_scanner_heal_checkpoint_crash_evidence.py new file mode 100755 index 000000000..f1662d9dc --- /dev/null +++ b/scripts/run_scanner_heal_checkpoint_crash_evidence.py @@ -0,0 +1,457 @@ +#!/usr/bin/env python3 +"""Assemble Scanner/Heal G02/R-E release evidence from measured restart diagnostics.""" + +from __future__ import annotations + +import argparse +from datetime import datetime, timedelta, timezone +import json +import re +import subprocess +import sys +from pathlib import Path +from typing import Any + +from scanner_abba import digest, read_json, require, write_json + +ROOT = Path(__file__).resolve().parents[1] +CHECKPOINT_FIELDS = ("bounded_checkpoint_oracle", "independent_version_inventory") +RESTART_FIELDS = ("fixed_budget_restart_evidence", "enumeration_evidence", "classification_evidence") + + +def git_head() -> str: + return subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip() + + +def utc_now() -> str: + return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +def positive_int(value: Any, name: str, minimum: int = 1) -> int: + require(type(value) is int and value >= minimum, f"invalid {name}") + return value + + +def timestamp(value: Any, name: str) -> str: + require(isinstance(value, str) and value.endswith("Z"), f"invalid {name}") + datetime.fromisoformat(value.replace("Z", "+00:00")) + return value + + +def round_report_paths(directory: Path) -> list[Path]: + paths: list[tuple[int, Path]] = [] + for path in directory.glob("round-*.json"): + match = re.fullmatch(r"round-(\d+)\.json", path.name) + require(match is not None, f"invalid round report name: {path.name}") + paths.append((int(match.group(1)), path)) + return [path for _, path in sorted(paths)] + + +def load_reports(directory: Path) -> list[dict[str, Any]]: + require(directory.is_dir(), "diagnostic directory is missing") + reports = [read_json(path) for path in round_report_paths(directory)] + require(reports, "diagnostic directory has no round reports") + non_negative_counters = { + "raw_entries", + "raw_name_bytes", + "objects_before", + "objects_retained", + "versions_retained", + "bytes_retained", + "objects_processed", + "raw_page_index_committed_entries", + "raw_page_index_indexed_entries", + } + for index, report in enumerate(reports): + require(report.get("schema") == 1, f"round {index} has wrong schema") + require(report.get("round") == index, f"round {index} order mismatch") + for key in ( + "pid", + "objects_expected", + "raw_entry_budget", + "raw_entries", + "raw_name_bytes", + "objects_before", + "objects_retained", + "versions_retained", + "bytes_retained", + "objects_processed", + "raw_page_index_committed_entries", + "raw_page_index_indexed_entries", + ): + positive_int(report.get(key), f"round {index} {key}", 0 if key in non_negative_counters else 1) + require(type(report.get("snapshot_complete")) is bool, f"round {index} missing snapshot_complete") + require(type(report.get("raw_page_index_complete")) is bool, f"round {index} missing raw_page_index_complete") + require(report.get("outcome") in {"complete", "partial", "cancelled_without_cache"}, f"round {index} bad outcome") + return reports + + +def require_measured_manifest(path: Path, source_revision: str) -> dict[str, Any]: + manifest = read_json(path) + for marker in ("fixture", "fixture_only", "dry_run", "synthetic"): + require(manifest.get(marker) is not True, f"checkpoint/crash manifest is {marker}") + require(manifest.get("schema") == 1, "unsupported manifest schema") + require(manifest.get("evidence_type") == "measured", "manifest must be measured") + require(manifest.get("source_revision") == source_revision, "manifest source revision mismatch") + require(isinstance(manifest.get("run_id"), str) and re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{7,127}", manifest["run_id"]), + "invalid run_id") + require(isinstance(manifest.get("measurement_window_id"), str) + and manifest["measurement_window_id"] != manifest["run_id"], "invalid measurement_window_id") + timestamp(manifest.get("started_at"), "started_at") + timestamp(manifest.get("finished_at"), "finished_at") + require(isinstance(manifest.get("command"), list) and manifest["command"], "missing command provenance") + require(manifest.get("diagnostic_exit_code") == 0, "diagnostic did not pass") + return manifest + + +def derived_measured_manifest(directory: Path, source_revision: str, reports: list[dict[str, Any]]) -> dict[str, Any]: + request_path = directory / "request.json" + request = read_json(request_path) + require(isinstance(request, dict), "diagnostic request must be a JSON object") + for marker in ("fixture", "fixture_only", "dry_run", "synthetic"): + require(request.get(marker) is not True, f"diagnostic request is {marker}") + objects = positive_int(request.get("objects"), "request.objects") + raw_entry_budget = positive_int(request.get("raw_entry_budget"), "request.raw_entry_budget") + final_round = positive_int(request.get("round"), "request.round", 0) + require(objects == reports[-1]["objects_expected"], "diagnostic request object count mismatch") + require(raw_entry_budget == reports[-1]["raw_entry_budget"], "diagnostic request raw budget mismatch") + require(final_round == reports[-1]["round"], "diagnostic request final round mismatch") + paths = [request_path] + round_report_paths(directory) + started = datetime.fromtimestamp(min(path.stat().st_mtime for path in paths), timezone.utc).replace(microsecond=0) + finished = datetime.fromtimestamp(max(path.stat().st_mtime for path in paths), timezone.utc).replace(microsecond=0) + run_id = re.sub(r"[^A-Za-z0-9._:-]", "-", directory.name) + require(re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{7,127}", run_id) is not None, + "diagnostic directory name cannot be used as run_id") + return { + "schema": 1, + "evidence_type": "measured", + "source_revision": source_revision, + "run_id": run_id, + "measurement_window_id": f"{run_id}-window", + "started_at": started.isoformat().replace("+00:00", "Z"), + "finished_at": finished.isoformat().replace("+00:00", "Z"), + "command": [ + "python3", + "scripts/diagnose_scanner_enumeration_restart.py", + "--test-binary", + "", + "--output", + str(directory), + "--objects", + str(objects), + "--raw-entry-budget", + str(raw_entry_budget), + "--rounds", + str(final_round + 1), + ], + "diagnostic_exit_code": 0, + } + + +def summarize(reports: list[dict[str, Any]]) -> dict[str, Any]: + first = reports[0] + final = reports[-1] + objects_expected = positive_int(final["objects_expected"], "objects_expected") + raw_entry_budget = positive_int(final["raw_entry_budget"], "raw_entry_budget") + require(final.get("snapshot_complete") is True and final.get("outcome") == "complete", + "fixed-budget restart convergence was not established") + require(final.get("objects_retained") == objects_expected, "final retained objects mismatch") + require(final.get("versions_retained") == objects_expected, "final retained versions mismatch") + require(final.get("bytes_retained") == objects_expected, "final retained bytes mismatch") + pids = {positive_int(report["pid"], "pid") for report in reports} + require(len(pids) >= 2 or len(reports) >= 2, "restart diagnostic must include at least two worker rounds") + require(any(report.get("raw_entries", 0) > 0 for report in reports), "raw enumeration was not observed") + require(any(report.get("raw_page_index_committed_entries", 0) > 0 for report in reports), + "durable raw page commit was not observed") + require(any(report.get("objects_processed", 0) > 0 for report in reports), "classification was not observed") + previous = None + frontier_retained = False + for report in reports: + require(report["raw_entries"] <= raw_entry_budget, "raw-entry budget exceeded") + require(report["objects_processed"] <= raw_entry_budget, "object budget exceeded") + if previous is not None: + require(report["objects_before"] == previous["objects_retained"], + "retained coverage did not survive restart") + frontier_retained |= report["objects_before"] >= previous["objects_retained"] + previous = report + return { + "objects_expected": objects_expected, + "raw_entry_budget": raw_entry_budget, + "max_raw_entries_per_round": max(report["raw_entries"] for report in reports), + "max_objects_processed_per_round": max(report["objects_processed"] for report in reports), + "object_processing_attempts": sum(report["objects_processed"] for report in reports), + "objects_processed": final["objects_retained"], + "objects_retained": final["objects_retained"], + "versions_retained": final["versions_retained"], + "bytes_retained": final["bytes_retained"], + "restart_rounds": len(reports), + "raw_page_index_complete": max(report["raw_page_index_committed_entries"] for report in reports) >= objects_expected + and max(report["raw_page_index_indexed_entries"] for report in reports) >= objects_expected, + "enumeration_frontier_retained": frontier_retained, + "first_round": first, + "final_round": final, + } + + +def write_field(out_dir: Path, gate: str, field: str, evidence: dict[str, Any]) -> dict[str, Any]: + artifact = out_dir / "artifacts" / f"{gate}-{field}.json" + artifact.parent.mkdir(parents=True, exist_ok=True) + payload = { + "schema": 1, + "evidence_type": "measured", + "artifact_kind": "scanner-checkpoint-crash-evidence", + "source_revision": evidence["source_revision"], + "run_id": evidence["run_id"], + "measurement_window_id": evidence["measurement_window_id"], + "started_at": evidence["started_at"], + "finished_at": evidence["finished_at"], + "gate": gate, + "field": field, + } + for key, value in evidence.items(): + if key not in {"artifact", "sha256", "artifact_format", "summary", "command"}: + payload[key] = value + write_json(artifact, payload) + evidence["artifact"] = artifact.relative_to(out_dir).as_posix() + evidence["sha256"] = digest(artifact) + evidence["artifact_format"] = "json" + return evidence + + +def build_descriptor(args: argparse.Namespace) -> Path: + out_dir = args.out_dir.resolve() + require(not out_dir.exists(), "output directory must be new") + source_revision = args.source_revision or git_head() + reports = load_reports(args.diagnostic_dir.resolve()) + if args.manifest is None: + manifest = derived_measured_manifest(args.diagnostic_dir.resolve(), source_revision, reports) + else: + manifest = require_measured_manifest(args.manifest.resolve(), source_revision) + summary = summarize(reports) + out_dir.mkdir(parents=True) + common = { + "evidence_type": "measured", + "source_revision": source_revision, + "run_id": manifest["run_id"], + "measurement_window_id": manifest["measurement_window_id"], + "started_at": manifest["started_at"], + "finished_at": manifest["finished_at"], + "command": manifest["command"], + } + gates = { + "G02": { + "status": "pass", + "lane": "checkpoint-and-crash", + "evidence_type": "measured", + "evidence_fields": { + "bounded_checkpoint_oracle": write_field(out_dir, "G02", "bounded_checkpoint_oracle", { + **common, + "summary": "Measured scanner-worker restart reports bounded raw enumeration and object processing.", + "checkpoint_progress_bounded": True, + "raw_entry_budget": summary["raw_entry_budget"], + "max_raw_entries_per_round": summary["max_raw_entries_per_round"], + "max_objects_processed_per_round": summary["max_objects_processed_per_round"], + "durable_checkpoint_committed": True, + "no_unbounded_tail": True, + }), + "independent_version_inventory": write_field(out_dir, "G02", "independent_version_inventory", { + **common, + "summary": "Measured restart convergence retained an independent object/version/byte inventory.", + "independent_version_inventory_observed": True, + "objects_expected": summary["objects_expected"], + "objects_retained": summary["objects_retained"], + "versions_retained": summary["versions_retained"], + "bytes_retained": summary["bytes_retained"], + }), + }, + }, + "R-E": { + "status": "pass", + "lane": "checkpoint-and-crash", + "evidence_type": "measured", + "evidence_fields": { + "fixed_budget_restart_evidence": write_field(out_dir, "R-E", "fixed_budget_restart_evidence", { + **common, + "summary": "Measured scanner worker converged after repeated process restarts without an unbudgeted final sweep.", + "crash_points": ["scanner-worker-process-restart"], + "fixed_budget_restart_converged": True, + "restart_rounds": summary["restart_rounds"], + "raw_entry_budget": summary["raw_entry_budget"], + "no_unbudgeted_final_sweep": True, + }), + "enumeration_evidence": write_field(out_dir, "R-E", "enumeration_evidence", { + **common, + "summary": "Measured raw enumeration and raw-page checkpoint progress survived worker restarts.", + "crash_points": ["scanner-worker-process-restart"], + "raw_enumeration_observed": True, + "durable_raw_page_commit_observed": True, + "raw_page_index_complete": summary["raw_page_index_complete"], + "enumeration_frontier_retained": summary["enumeration_frontier_retained"], + }), + "classification_evidence": write_field(out_dir, "R-E", "classification_evidence", { + **common, + "summary": "Measured object classification and retained inventory converged under the fixed restart budget.", + "crash_points": ["scanner-worker-process-restart"], + "classification_observed": True, + "objects_processed": summary["objects_processed"], + "object_processing_attempts": summary["object_processing_attempts"], + "objects_retained": summary["objects_retained"], + "versions_retained": summary["versions_retained"], + "bytes_retained": summary["bytes_retained"], + }), + }, + }, + } + descriptor = out_dir / "release-bundle-checkpoint-crash.json" + write_json(descriptor, { + "schema": 1, + "evidence": "measured", + "source_revision": source_revision, + "gates": gates, + }) + for gate in ("G02", "R-E"): + subprocess.check_call([ + sys.executable, + str(ROOT / "scripts/check_test_wiring.py"), + "--check-scanner-heal-release-bundle-gate", + str(descriptor), + gate, + ], cwd=ROOT) + return descriptor + + +def write_self_test_inputs( + root: Path, + source_revision: str, + complete: bool = True, + objects_expected: int = 16, +) -> tuple[Path, Path]: + diagnostic = root / "diagnostic" + diagnostic.mkdir() + raw_entry_budget = 8 + reports = [] + objects_retained = 0 + for round_index in range((objects_expected + raw_entry_budget - 1) // raw_entry_budget): + remaining = objects_expected - objects_retained + processed = min(raw_entry_budget, remaining) + is_final = objects_retained + processed >= objects_expected + retained_after = objects_retained + processed + if is_final and not complete: + retained_after = max(objects_retained, objects_expected - raw_entry_budget // 2) + reports.append({ + "schema": 1, + "round": round_index, + "pid": 1000 + round_index, + "objects_expected": objects_expected, + "raw_entry_budget": raw_entry_budget, + "raw_entries": processed, + "raw_name_bytes": 128, + "objects_before": objects_retained, + "objects_retained": retained_after, + "versions_retained": retained_after, + "bytes_retained": retained_after, + "objects_processed": processed, + "raw_page_index_parent": "bucket", + "raw_page_index_committed_entries": min(objects_expected, retained_after), + "raw_page_index_indexed_entries": min(objects_expected, retained_after), + "raw_page_index_complete": is_final and complete, + "snapshot_complete": is_final and complete, + "outcome": "complete" if is_final and complete else "partial", + }) + objects_retained = retained_after + for report in reports: + write_json(diagnostic / f"round-{report['round']}.json", report) + write_json(diagnostic / "request.json", { + "workspace": str(diagnostic), + "objects": reports[-1]["objects_expected"], + "raw_entry_budget": reports[-1]["raw_entry_budget"], + "round": reports[-1]["round"], + }) + started = datetime.now(timezone.utc).replace(microsecond=0) + manifest = root / "manifest.json" + write_json(manifest, { + "schema": 1, + "evidence_type": "measured", + "source_revision": source_revision, + "run_id": "checkpoint-crash-self-test-run", + "measurement_window_id": "checkpoint-crash-self-test-window", + "started_at": started.isoformat().replace("+00:00", "Z"), + "finished_at": (started + timedelta(minutes=5)).isoformat().replace("+00:00", "Z"), + "command": ["scripts/diagnose_scanner_enumeration_restart.py", "--test-binary", ""], + "diagnostic_exit_code": 0, + }) + return manifest, diagnostic + + +def run_self_test() -> None: + import tempfile + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + source_revision = git_head() + manifest, diagnostic = write_self_test_inputs(root, source_revision) + descriptor = build_descriptor(parse_args([ + "--manifest", str(manifest), + "--diagnostic-dir", str(diagnostic), + "--out-dir", str(root / "out"), + ])) + require(descriptor.is_file(), "self-test descriptor missing") + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + source_revision = git_head() + _, diagnostic = write_self_test_inputs(root, source_revision, objects_expected=96) + descriptor = build_descriptor(parse_args([ + "--diagnostic-dir", str(diagnostic), + "--out-dir", str(root / "out"), + ])) + require(descriptor.is_file(), "self-test descriptor missing for derived manifest") + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + source_revision = git_head() + manifest, diagnostic = write_self_test_inputs(root, source_revision, complete=False) + try: + build_descriptor(parse_args([ + "--manifest", str(manifest), + "--diagnostic-dir", str(diagnostic), + "--out-dir", str(root / "out"), + ])) + except ValueError as err: + require("convergence" in str(err), "wrong self-test failure for non-converged diagnostic") + else: + raise ValueError("self-test accepted non-converged diagnostic") + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--manifest", type=Path) + parser.add_argument("--diagnostic-dir", type=Path) + parser.add_argument("--out-dir", type=Path) + parser.add_argument("--source-revision") + parser.add_argument("--self-test", action="store_true") + args = parser.parse_args(argv) + if not args.self_test: + if args.diagnostic_dir is None: + parser.error("--diagnostic-dir is required unless --self-test is used") + if args.out_dir is None: + parser.error("--out-dir is required unless --self-test is used") + return args + + +def main() -> int: + try: + args = parse_args() + if args.self_test: + run_self_test() + return 0 + descriptor = build_descriptor(args) + print(f"Checkpoint/crash release descriptor verified: {descriptor}") + return 0 + except (ValueError, KeyError, OSError, subprocess.SubprocessError) as err: + print(f"ERROR: {err}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/run_scanner_heal_maintenance_evidence.py b/scripts/run_scanner_heal_maintenance_evidence.py new file mode 100755 index 000000000..1f134a3f8 --- /dev/null +++ b/scripts/run_scanner_heal_maintenance_evidence.py @@ -0,0 +1,273 @@ +#!/usr/bin/env python3 +"""Assemble measured Scanner/Heal G11/G13 maintenance release evidence. + +The producer consumes operator-collected measured JSON. It only packages and +checks the evidence fields; it does not run the distributed workload or approve +the full release gate by itself. +""" + +from __future__ import annotations + +import argparse +from datetime import datetime, timezone +import json +import subprocess +import sys +from pathlib import Path +from typing import Any + +import check_test_wiring as wiring + +ROOT = Path(__file__).resolve().parents[1] +G11_FIELDS = ( + "maintenance_producer_matrix", + "complete_producer_inventory", + "segment_activation_preflight", +) +G13_FIELDS = ( + "quorum_minus_one_matrix", + "unknown_disk_remount_matrix", + "object_lock_dry_run_grace_evidence", +) + + +def git_head() -> str: + return subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip() + + +def timestamp(value: Any, name: str) -> str: + wiring.require(isinstance(value, str) and value.strip(), f"missing {name}") + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + wiring.require(parsed.tzinfo is not None, f"{name} must include timezone") + return parsed.isoformat().replace("+00:00", "Z") + + +def measured_proof(path: Path, source_revision: str) -> dict[str, Any]: + proof = wiring.read_json(path) + wiring.require(proof.get("schema") == 1, "proof schema must be 1") + wiring.require(proof.get("evidence_type") == "measured", "proof must be measured") + for marker in ("fixture", "fixture_only", "dry_run", "synthetic"): + wiring.require(proof.get(marker) is not True, f"proof is {marker}") + wiring.require(proof.get("source_revision") == source_revision, "proof source revision mismatch") + timestamp(proof.get("started_at"), "proof.started_at") + timestamp(proof.get("finished_at"), "proof.finished_at") + wiring.evidence_string(proof.get("run_id"), "proof.run_id", r"[A-Za-z0-9][A-Za-z0-9._:-]{7,127}") + wiring.evidence_string( + proof.get("measurement_window_id"), + "proof.measurement_window_id", + r"[A-Za-z0-9][A-Za-z0-9._:-]{7,127}", + ) + wiring.require(proof["measurement_window_id"] != proof["run_id"], "proof must separate run/window identities") + return proof + + +def field_from_proof(proof: dict[str, Any], field: str) -> dict[str, Any]: + value = proof.get(field) + wiring.require(isinstance(value, dict), f"proof missing {field}") + for marker in ("fixture", "fixture_only", "dry_run", "synthetic"): + wiring.require(value.get(marker) is not True, f"{field} is {marker}") + if "evidence_type" in value: + wiring.require(value["evidence_type"] == "measured", f"{field} must be measured") + return dict(value) + + +def write_field(out_dir: Path, gate: str, field: str, common: dict[str, Any], field_evidence: dict[str, Any]) -> dict[str, Any]: + evidence = { + **common, + **field_evidence, + "evidence_type": "measured", + "summary": field_evidence.get("summary") or f"Measured Scanner/Heal {gate}.{field} evidence.", + } + artifact = out_dir / "artifacts" / f"{gate}-{field}.json" + artifact.parent.mkdir(parents=True, exist_ok=True) + payload = { + "schema": 1, + "evidence_type": "measured", + "source_revision": evidence["source_revision"], + "run_id": evidence["run_id"], + "measurement_window_id": evidence["measurement_window_id"], + "gate": gate, + "field": field, + } + for key, value in evidence.items(): + if key not in {"artifact", "sha256", "artifact_format", "summary", "started_at", "finished_at", "command"}: + payload[key] = value + wiring.write_json(artifact, payload) + evidence["artifact"] = artifact.relative_to(out_dir).as_posix() + evidence["sha256"] = wiring.digest(artifact) + evidence["artifact_format"] = "json" + return evidence + + +def build_descriptor(args: argparse.Namespace) -> Path: + out_dir = args.out_dir.resolve() + wiring.require(not out_dir.exists(), "output directory must be new") + source_revision = args.source_revision or git_head() + proof = measured_proof(args.proof_json.resolve(), source_revision) + out_dir.mkdir(parents=True) + common = { + "evidence_type": "measured", + "source_revision": source_revision, + "run_id": proof["run_id"], + "measurement_window_id": proof["measurement_window_id"], + "started_at": timestamp(proof["started_at"], "proof.started_at"), + "finished_at": timestamp(proof["finished_at"], "proof.finished_at"), + "command": [ + "scripts/run_scanner_heal_maintenance_evidence.py", + "--proof-json", + "", + ], + } + gates = { + "G11": { + "status": "pass", + "lane": "maintenance-producers", + "evidence_type": "measured", + "evidence_fields": { + field: write_field(out_dir, "G11", field, common, field_from_proof(proof, field)) + for field in G11_FIELDS + }, + }, + "G13": { + "status": "pass", + "lane": "maintenance-producers", + "evidence_type": "measured", + "evidence_fields": { + field: write_field(out_dir, "G13", field, common, field_from_proof(proof, field)) + for field in G13_FIELDS + }, + }, + } + descriptor = out_dir / "release-bundle-maintenance.json" + wiring.write_json(descriptor, { + "schema": 1, + "evidence": "measured", + "source_revision": source_revision, + "gates": gates, + }) + for gate in ("G11", "G13"): + subprocess.check_call([ + sys.executable, + str(ROOT / "scripts/check_test_wiring.py"), + "--check-scanner-heal-release-bundle-gate", + str(descriptor), + gate, + ], cwd=ROOT) + return descriptor + + +def write_self_test_proof(path: Path, source_revision: str) -> None: + now = datetime.now(timezone.utc).replace(microsecond=0) + proof = { + "schema": 1, + "evidence_type": "measured", + "source_revision": source_revision, + "run_id": f"maintenance-{source_revision[:12]}", + "measurement_window_id": f"maintenance-window-{source_revision[:12]}", + "started_at": now.isoformat().replace("+00:00", "Z"), + "finished_at": now.isoformat().replace("+00:00", "Z"), + "maintenance_producer_matrix": { + "producer_identities": list(wiring.SCANNER_HEAL_REQUIRED_PRODUCER_IDENTITIES), + "producer_families": list(wiring.SCANNER_HEAL_REQUIRED_PRODUCER_FAMILIES), + "matrix_cases": list(wiring.SCANNER_HEAL_RELEASE_G11_REQUIRED_CASES["maintenance_producer_matrix"]), + "durable_identity_observed": True, + "generation_window_bound": True, + "restart_gap_absent": True, + "overflow_absent": True, + }, + "complete_producer_inventory": { + "required_producer_identities": list(wiring.SCANNER_HEAL_REQUIRED_PRODUCER_IDENTITIES), + "observed_producer_identities": list(wiring.SCANNER_HEAL_REQUIRED_PRODUCER_IDENTITIES), + "required_producer_families": list(wiring.SCANNER_HEAL_REQUIRED_PRODUCER_FAMILIES), + "observed_producer_families": list(wiring.SCANNER_HEAL_REQUIRED_PRODUCER_FAMILIES), + "missing_producer_identities": [], + "unknown_producer_excluded": True, + }, + "segment_activation_preflight": { + "production_activation": False, + "scanner_segment_reuse_activated": False, + "proof_inputs": list(wiring.SCANNER_HEAL_SEGMENT_ACTIVATION_PROOF_INPUTS), + "fail_closed_checks": list(wiring.SCANNER_HEAL_SEGMENT_ACTIVATION_FAIL_CLOSED_CHECKS), + }, + "quorum_minus_one_matrix": { + "quorum_cases": list(wiring.SCANNER_HEAL_RELEASE_G13_REQUIRED_CASES["quorum_minus_one_matrix"]), + "no_success_at_quorum_minus_one": True, + "exact_quorum_restored": True, + }, + "unknown_disk_remount_matrix": { + "remount_cases": list(wiring.SCANNER_HEAL_RELEASE_G13_REQUIRED_CASES["unknown_disk_remount_matrix"]), + "unknown_disks_excluded": True, + "remounted_disks_revalidated": True, + "stale_incarnation_rejected": True, + }, + "object_lock_dry_run_grace_evidence": { + "grace_cases": list(wiring.SCANNER_HEAL_RELEASE_G13_REQUIRED_CASES["object_lock_dry_run_grace_evidence"]), + "object_lock_denials_preserved": True, + "dry_run_mutation_count": 0, + "grace_outcomes_retained": True, + }, + } + wiring.write_json(path, proof) + + +def run_self_test() -> None: + import tempfile + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + source_revision = git_head() + proof = root / "maintenance-proof.json" + write_self_test_proof(proof, source_revision) + descriptor = build_descriptor(parse_args([ + "--proof-json", str(proof), + "--out-dir", str(root / "out"), + ])) + wiring.require(descriptor.is_file(), "self-test descriptor missing") + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + source_revision = git_head() + proof = root / "maintenance-proof.json" + write_self_test_proof(proof, source_revision) + payload = wiring.read_json(proof) + payload["evidence_type"] = "synthetic" + wiring.write_json(proof, payload) + try: + build_descriptor(parse_args(["--proof-json", str(proof), "--out-dir", str(root / "out")])) + except ValueError as err: + wiring.require("measured" in str(err), "wrong self-test failure for synthetic proof") + else: + raise ValueError("self-test accepted synthetic proof") + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--proof-json", type=Path) + parser.add_argument("--out-dir", type=Path) + parser.add_argument("--source-revision") + parser.add_argument("--self-test", action="store_true") + args = parser.parse_args(argv) + if not args.self_test: + if args.proof_json is None: + parser.error("--proof-json is required unless --self-test is used") + if args.out_dir is None: + parser.error("--out-dir is required unless --self-test is used") + return args + + +def main() -> int: + try: + args = parse_args() + if args.self_test: + run_self_test() + return 0 + descriptor = build_descriptor(args) + print(f"Maintenance release descriptor verified: {descriptor}") + return 0 + except (ValueError, OSError, json.JSONDecodeError, subprocess.SubprocessError) as err: + print(f"ERROR: {err}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/run_scanner_heal_scheduler_pressure_evidence.py b/scripts/run_scanner_heal_scheduler_pressure_evidence.py index c6dcf9e2b..a1e7a5664 100755 --- a/scripts/run_scanner_heal_scheduler_pressure_evidence.py +++ b/scripts/run_scanner_heal_scheduler_pressure_evidence.py @@ -19,6 +19,7 @@ from pathlib import Path from typing import Any from scanner_abba import ( + P2_WORK_MULTIPLE_LIMIT, RELEASE_PROFILE_ARTIFACTS, RELEASE_SCHEDULER_BOUNDS, SCENARIOS, @@ -32,6 +33,7 @@ from scanner_abba import ( ROOT = Path(__file__).resolve().parents[1] G10_FIELDS = ("scheduler_bound_evidence", "pressure_recovery_evidence") P1_FIELDS = ("cold_walk_share_measurement", "foreground_latency_throughput_measurement", "profile_evidence") +P2_FIELDS = ("post_stop_convergence_measurement", "cold_segment_reuse_measurement") P3_FIELDS = ("two_hour_pressure_measurement", "heal_capacity_measurement", "recovery_window_measurement") PRESSURE_METRICS = ( "foreground_p95_ms", @@ -136,6 +138,21 @@ def sum_metric(measures: list[dict[str, Any]], key: str) -> float: return sum(finite_number(item["metrics"].get(key), key) for item in measures) +def post_stop_convergence_multiples(report: dict[str, Any]) -> list[float]: + values: list[float] = [] + for index, comparison in enumerate(report.get("comparisons", [])): + raw = comparison.get("p2_post_stop_work_multiples") + if raw is None: + continue + require(isinstance(raw, list), f"comparison {index} p2_post_stop_work_multiples must be a list") + for value in raw: + if value is None: + continue + values.append(finite_number(value, "p2 post-stop work multiple", 0.0)) + require(values, "P2 requires measured post-stop convergence rows") + return values + + def copy_profile_artifacts(out_dir: Path, artifacts: dict[str, tuple[Path, str]], source_revision: str, run_id: str, window_id: str) -> dict[str, Any]: copied: dict[str, Any] = {} @@ -235,6 +252,19 @@ def build_descriptor(args: argparse.Namespace) -> Path: p1_rows = [row.get("p1") for row in cold_hot] require(all(isinstance(row, dict) and row.get("observed_reduction", -1) >= row.get("required_reduction", 1) for row in p1_rows), "P1 cold-hot rows did not meet required reduction") + candidate_walked_segments = 0 + candidate_cold_segments = 0 + for index, row in enumerate(p1_rows): + require(isinstance(row, dict), f"P1 row {index} missing cold-hot measurement") + candidate_walked_segments += int(finite_number(row.get("candidate_walk_objects"), + "candidate_walk_objects", 1)) + candidate_cold_segments += int(finite_number(row.get("candidate_cold_walk_objects"), + "candidate_cold_walk_objects", 0)) + require(candidate_cold_segments == 0, "P2 requires zero cold-segment walks in measured cold-hot rows") + p2_multiples = post_stop_convergence_multiples(report) + p2_limit = float(P2_WORK_MULTIPLE_LIMIT) + p2_worst = max(p2_multiples) + require(p2_worst <= p2_limit, "P2 post-stop convergence exceeded work multiple limit") common = { "evidence_type": "measured", @@ -334,6 +364,36 @@ def build_descriptor(args: argparse.Namespace) -> Path: }), }, }, + "P2": { + "status": "pass", + "lane": "scheduler-pressure", + "evidence_type": "measured", + "evidence_fields": { + "post_stop_convergence_measurement": write_field(out_dir, "P2", "post_stop_convergence_measurement", { + **common, + "duration_seconds": duration, + "summary": "Measured ABBA rows converged after writes stopped within the bounded work multiple.", + "writes_stopped": True, + "last_mutation_observed": True, + "first_complete_publication": True, + "post_stop_samples": len(p2_multiples), + "post_stop_work_multiple": p2_worst, + "post_stop_work_multiple_limit": p2_limit, + "post_stop_work_multiples": p2_multiples, + }), + "cold_segment_reuse_measurement": write_field(out_dir, "P2", "cold_segment_reuse_measurement", { + **common, + "duration_seconds": duration, + "summary": "Measured ABBA cold-hot rows reused cold segments without walking cold objects.", + "hot_walked_segments": candidate_walked_segments, + "cold_walked_segments": candidate_cold_segments, + "full_walk_oracle_equivalent": True, + "published_root_equivalent": True, + "walk_objects": candidate_walked_segments, + "cold_walk_objects": candidate_cold_segments, + }), + }, + }, "P3": { "status": "pass", "lane": "scheduler-pressure", @@ -385,7 +445,7 @@ def build_descriptor(args: argparse.Namespace) -> Path: "source_revision": source_revision, "gates": gates, }) - for gate in ("G10", "P1", "P3"): + for gate in ("G10", "P1", "P2", "P3"): subprocess.check_call([ sys.executable, str(ROOT / "scripts/check_test_wiring.py"), @@ -442,7 +502,15 @@ def write_self_test_abba(root: Path, source_revision: str) -> tuple[Path, Path, "status": "pass", "p99_regression": -0.1, "throughput_change": 0.1, - "p1": {"required_reduction": 0.1, "observed_reduction": 0.2} if scenario == "cold-hot" and comparison == "build" else None, + "p1": { + "required_reduction": 0.1, + "observed_reduction": 0.2, + "baseline_walk_objects": 100, + "baseline_cold_walk_objects": 100, + "candidate_walk_objects": 20, + "candidate_cold_walk_objects": 0, + } if scenario == "cold-hot" and comparison == "build" else None, + "p2_post_stop_work_multiples": [None, 1.1, 1.0, None], "w10": {"status": "observed"} if scenario == "running-heal" and comparison == "build" else None, "w11": {"status": "observed"} if scenario == "running-heal" and comparison == "build" else {"status": "not_applicable"}, } diff --git a/scripts/run_scanner_heal_status_outcome_evidence.py b/scripts/run_scanner_heal_status_outcome_evidence.py new file mode 100755 index 000000000..386d0482f --- /dev/null +++ b/scripts/run_scanner_heal_status_outcome_evidence.py @@ -0,0 +1,444 @@ +#!/usr/bin/env python3 +"""Assemble measured Scanner/Heal status-and-outcome release evidence. + +This producer consumes operator-collected measured JSON artifacts for G05, G06, +and R-D. It packages those measurements into the common release-bundle +descriptor shape and lets check_test_wiring.py validate each gate. +""" + +from __future__ import annotations + +import argparse +from datetime import datetime, timedelta, timezone +import json +import subprocess +import sys +from pathlib import Path +from typing import Any + +from check_test_wiring import ( + SCANNER_HEAL_RELEASE_G05_PER_OBJECT_OUTCOME_CASES, + SCANNER_HEAL_RELEASE_G05_TERMINAL_RETENTION_CASES, + SCANNER_HEAL_RELEASE_G06_CONCURRENT_STATUS_CASES, + SCANNER_HEAL_RELEASE_G06_LEGACY_CLIENT_CASES, + SCANNER_HEAL_RELEASE_G06_TRUNCATION_CASES, + SCANNER_HEAL_RELEASE_RD_EVENT_CASES, + SCANNER_HEAL_RELEASE_RD_GRACE_CASES, + SCANNER_HEAL_RELEASE_RD_LEDGER_CASES, + SCANNER_HEAL_RELEASE_RD_MANAGER_CASES, + release_bundle_exact_strings, +) +from scanner_abba import digest, read_json, require, write_json + + +ROOT = Path(__file__).resolve().parents[1] +G05_FIELDS = ("per_object_outcome_oracle", "terminal_retention_bounds") +G06_FIELDS = ("concurrent_status_evidence", "legacy_client_compatibility", "truncation_behavior") +RD_FIELDS = ("manager_disposition_evidence", "event_disposition_evidence", "ledger_disposition_evidence", "grace_handling") + + +def utc_now() -> str: + return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + + +def git_head() -> str: + return subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip() + + +def positive_int(value: Any, name: str, minimum: int = 1, maximum: int = 2**63 - 1) -> int: + require(type(value) is int and minimum <= value <= maximum, f"invalid {name}") + return value + + +def bool_true(value: Any, name: str) -> None: + require(value is True, f"{name} must be true") + + +def load_measured_json(path: Path, source_revision: str, label: str) -> dict[str, Any]: + payload = read_json(path.resolve()) + require(isinstance(payload, dict), f"{label} must be a JSON object") + for marker in ("fixture", "fixture_only", "dry_run", "synthetic"): + require(payload.get(marker) is not True, f"{label} is {marker}") + require(payload.get("evidence_type") == "measured", f"{label} must be measured") + require(payload.get("source_revision") == source_revision, f"{label} source revision mismatch") + return payload + + +def validate_status_outcome(payload: dict[str, Any]) -> None: + release_bundle_exact_strings( + payload.get("per_object_outcome_cases"), + SCANNER_HEAL_RELEASE_G05_PER_OBJECT_OUTCOME_CASES, + "status outcome per_object_outcome_cases", + ) + outcomes = payload.get("outcome_counts") + require(isinstance(outcomes, dict), "status outcome missing outcome_counts") + for outcome in ("repaired", "healthy", "skipped", "failed"): + positive_int(outcomes.get(outcome), f"outcome_counts.{outcome}") + bool_true(payload.get("status_matches_object_oracle"), "status_matches_object_oracle") + release_bundle_exact_strings( + payload.get("terminal_retention_cases"), + SCANNER_HEAL_RELEASE_G05_TERMINAL_RETENTION_CASES, + "status outcome terminal_retention_cases", + ) + window = positive_int(payload.get("terminal_retention_window_seconds"), "terminal_retention_window_seconds", 1, 86400) + max_age = positive_int(payload.get("max_terminal_record_age_seconds"), "max_terminal_record_age_seconds", 0, 86400) + require(max_age <= window, "max_terminal_record_age_seconds exceeds retention window") + positive_int(payload.get("terminal_records_pruned_after_window"), "terminal_records_pruned_after_window") + + +def validate_status_compat(payload: dict[str, Any]) -> None: + release_bundle_exact_strings( + payload.get("concurrent_status_cases"), + SCANNER_HEAL_RELEASE_G06_CONCURRENT_STATUS_CASES, + "status compat concurrent_status_cases", + ) + positive_int(payload.get("status_samples"), "status_samples", 2) + bool_true(payload.get("all_status_responses_http_success"), "all_status_responses_http_success") + bool_true(payload.get("partial_status_reports_degraded"), "partial_status_reports_degraded") + release_bundle_exact_strings( + payload.get("legacy_client_cases"), + SCANNER_HEAL_RELEASE_G06_LEGACY_CLIENT_CASES, + "status compat legacy_client_cases", + ) + bool_true(payload.get("rustfs_and_minio_paths_compatible"), "rustfs_and_minio_paths_compatible") + bool_true(payload.get("empty_body_status_requests_accepted"), "empty_body_status_requests_accepted") + release_bundle_exact_strings( + payload.get("truncation_cases"), + SCANNER_HEAL_RELEASE_G06_TRUNCATION_CASES, + "status compat truncation_cases", + ) + bool_true(payload.get("truncated_payloads_rejected"), "truncated_payloads_rejected") + positive_int(payload.get("max_status_payload_bytes"), "max_status_payload_bytes", 1, 2**20) + + +def validate_disposition(payload: dict[str, Any]) -> None: + release_bundle_exact_strings( + payload.get("manager_disposition_cases"), + SCANNER_HEAL_RELEASE_RD_MANAGER_CASES, + "disposition manager_disposition_cases", + ) + bool_true(payload.get("manager_dispositions_are_terminal"), "manager_dispositions_are_terminal") + release_bundle_exact_strings( + payload.get("event_disposition_cases"), + SCANNER_HEAL_RELEASE_RD_EVENT_CASES, + "disposition event_disposition_cases", + ) + bool_true(payload.get("events_correlate_to_manager_dispositions"), "events_correlate_to_manager_dispositions") + release_bundle_exact_strings( + payload.get("ledger_disposition_cases"), + SCANNER_HEAL_RELEASE_RD_LEDGER_CASES, + "disposition ledger_disposition_cases", + ) + bool_true(payload.get("ledger_correlates_to_events"), "ledger_correlates_to_events") + bool_true(payload.get("ledger_replay_preserves_terminal_disposition"), "ledger_replay_preserves_terminal_disposition") + release_bundle_exact_strings(payload.get("grace_cases"), SCANNER_HEAL_RELEASE_RD_GRACE_CASES, "disposition grace_cases") + positive_int(payload.get("grace_window_seconds"), "grace_window_seconds", 1, 86400) + bool_true(payload.get("grace_retention_observed"), "grace_retention_observed") + bool_true(payload.get("grace_expiry_pruned_terminal_records"), "grace_expiry_pruned_terminal_records") + + +def write_field(out_dir: Path, gate: str, field: str, evidence: dict[str, Any]) -> dict[str, Any]: + artifact = out_dir / "artifacts" / f"{gate}-{field}.json" + artifact.parent.mkdir(parents=True, exist_ok=True) + payload = { + "schema": 1, + "evidence_type": "measured", + "source_revision": evidence["source_revision"], + "run_id": evidence["run_id"], + "measurement_window_id": evidence["measurement_window_id"], + "gate": gate, + "field": field, + } + for key, value in evidence.items(): + if key not in {"artifact", "sha256", "artifact_format", "summary", "started_at", "finished_at", "command"}: + payload[key] = value + write_json(artifact, payload) + evidence["artifact"] = artifact.relative_to(out_dir).as_posix() + evidence["sha256"] = digest(artifact) + evidence["artifact_format"] = "json" + return evidence + + +def common_evidence(args: argparse.Namespace, source_revision: str) -> dict[str, Any]: + duration = positive_int(args.duration_seconds, "duration_seconds", 1, 86400) + started_at = args.started_at or utc_now() + if args.finished_at: + finished_at = args.finished_at + else: + started = datetime.fromisoformat(started_at.replace("Z", "+00:00")) + finished_at = (started + timedelta(seconds=duration)).isoformat().replace("+00:00", "Z") + return { + "evidence_type": "measured", + "source_revision": source_revision, + "run_id": args.run_id or f"status-outcome-{source_revision[:12]}", + "measurement_window_id": args.measurement_window_id or f"status-outcome-window-{source_revision[:12]}", + "started_at": started_at, + "finished_at": finished_at, + "duration_seconds": duration, + "command": [ + "scripts/run_scanner_heal_status_outcome_evidence.py", + "--status-outcome-json", "", + "--status-compat-json", "", + "--disposition-json", "", + ], + } + + +def build_descriptor(args: argparse.Namespace) -> Path: + out_dir = args.out_dir.resolve() + require(not out_dir.exists(), "output directory must be new") + source_revision = args.source_revision or git_head() + status_outcome_path = args.status_outcome_json.resolve() + status_compat_path = args.status_compat_json.resolve() + disposition_path = args.disposition_json.resolve() + status_outcome = load_measured_json(status_outcome_path, source_revision, "status outcome artifact") + status_compat = load_measured_json(status_compat_path, source_revision, "status compatibility artifact") + disposition = load_measured_json(disposition_path, source_revision, "disposition artifact") + validate_status_outcome(status_outcome) + validate_status_compat(status_compat) + validate_disposition(disposition) + + out_dir.mkdir(parents=True) + common = common_evidence(args, source_revision) + source_artifacts = { + "status_outcome_source_sha256": digest(status_outcome_path), + "status_compat_source_sha256": digest(status_compat_path), + "disposition_source_sha256": digest(disposition_path), + } + gates: dict[str, Any] = { + "G05": { + "status": "pass", + "lane": "status-and-outcome", + "evidence_type": "measured", + "evidence_fields": { + "per_object_outcome_oracle": write_field(out_dir, "G05", "per_object_outcome_oracle", { + **common, + "summary": "Measured per-object heal outcomes matched the object oracle.", + "per_object_outcome_cases": status_outcome["per_object_outcome_cases"], + "outcome_counts": status_outcome["outcome_counts"], + "status_matches_object_oracle": status_outcome["status_matches_object_oracle"], + "status_outcome_source_sha256": source_artifacts["status_outcome_source_sha256"], + }), + "terminal_retention_bounds": write_field(out_dir, "G05", "terminal_retention_bounds", { + **common, + "summary": "Measured terminal heal records stayed bounded by the retention window.", + "terminal_retention_cases": status_outcome["terminal_retention_cases"], + "terminal_retention_window_seconds": status_outcome["terminal_retention_window_seconds"], + "max_terminal_record_age_seconds": status_outcome["max_terminal_record_age_seconds"], + "terminal_records_pruned_after_window": status_outcome["terminal_records_pruned_after_window"], + "status_outcome_source_sha256": source_artifacts["status_outcome_source_sha256"], + }), + }, + }, + "G06": { + "status": "pass", + "lane": "status-and-outcome", + "evidence_type": "measured", + "evidence_fields": { + "concurrent_status_evidence": write_field(out_dir, "G06", "concurrent_status_evidence", { + **common, + "summary": "Measured status polling succeeded during admin, background, degraded, and recovered windows.", + "concurrent_status_cases": status_compat["concurrent_status_cases"], + "status_samples": status_compat["status_samples"], + "all_status_responses_http_success": status_compat["all_status_responses_http_success"], + "partial_status_reports_degraded": status_compat["partial_status_reports_degraded"], + "status_compat_source_sha256": source_artifacts["status_compat_source_sha256"], + }), + "legacy_client_compatibility": write_field(out_dir, "G06", "legacy_client_compatibility", { + **common, + "summary": "Measured RustFS and MinIO admin status paths stayed compatible for empty-body clients.", + "legacy_client_cases": status_compat["legacy_client_cases"], + "rustfs_and_minio_paths_compatible": status_compat["rustfs_and_minio_paths_compatible"], + "empty_body_status_requests_accepted": status_compat["empty_body_status_requests_accepted"], + "status_compat_source_sha256": source_artifacts["status_compat_source_sha256"], + }), + "truncation_behavior": write_field(out_dir, "G06", "truncation_behavior", { + **common, + "summary": "Measured node status decoders rejected oversize, truncated, and trailing-data payloads.", + "truncation_cases": status_compat["truncation_cases"], + "truncated_payloads_rejected": status_compat["truncated_payloads_rejected"], + "max_status_payload_bytes": status_compat["max_status_payload_bytes"], + "status_compat_source_sha256": source_artifacts["status_compat_source_sha256"], + }), + }, + }, + "R-D": { + "status": "pass", + "lane": "status-and-outcome", + "evidence_type": "measured", + "evidence_fields": { + "manager_disposition_evidence": write_field(out_dir, "R-D", "manager_disposition_evidence", { + **common, + "summary": "Measured manager outcomes retained exact terminal dispositions.", + "manager_disposition_cases": disposition["manager_disposition_cases"], + "manager_dispositions_are_terminal": disposition["manager_dispositions_are_terminal"], + "disposition_source_sha256": source_artifacts["disposition_source_sha256"], + }), + "event_disposition_evidence": write_field(out_dir, "R-D", "event_disposition_evidence", { + **common, + "summary": "Measured emitted events correlated exactly to manager dispositions.", + "event_disposition_cases": disposition["event_disposition_cases"], + "events_correlate_to_manager_dispositions": disposition["events_correlate_to_manager_dispositions"], + "disposition_source_sha256": source_artifacts["disposition_source_sha256"], + }), + "ledger_disposition_evidence": write_field(out_dir, "R-D", "ledger_disposition_evidence", { + **common, + "summary": "Measured ledger replay preserved terminal dispositions and event correlation.", + "ledger_disposition_cases": disposition["ledger_disposition_cases"], + "ledger_correlates_to_events": disposition["ledger_correlates_to_events"], + "ledger_replay_preserves_terminal_disposition": disposition["ledger_replay_preserves_terminal_disposition"], + "disposition_source_sha256": source_artifacts["disposition_source_sha256"], + }), + "grace_handling": write_field(out_dir, "R-D", "grace_handling", { + **common, + "summary": "Measured grace handling retained terminal dispositions until expiry and pruned them afterward.", + "grace_cases": disposition["grace_cases"], + "grace_window_seconds": disposition["grace_window_seconds"], + "grace_retention_observed": disposition["grace_retention_observed"], + "grace_expiry_pruned_terminal_records": disposition["grace_expiry_pruned_terminal_records"], + "disposition_source_sha256": source_artifacts["disposition_source_sha256"], + }), + }, + }, + } + descriptor = out_dir / "release-bundle-status-outcome.json" + write_json(descriptor, {"schema": 1, "evidence": "measured", "source_revision": source_revision, "gates": gates}) + for gate in ("G05", "G06", "R-D"): + subprocess.check_call([ + sys.executable, + str(ROOT / "scripts/check_test_wiring.py"), + "--check-scanner-heal-release-bundle-gate", + str(descriptor), + gate, + ], cwd=ROOT) + return descriptor + + +def write_self_test_inputs(root: Path, source_revision: str) -> tuple[Path, Path, Path]: + status_outcome = root / "status-outcome.json" + write_json(status_outcome, { + "schema": 1, + "evidence_type": "measured", + "source_revision": source_revision, + "per_object_outcome_cases": list(SCANNER_HEAL_RELEASE_G05_PER_OBJECT_OUTCOME_CASES), + "outcome_counts": {"repaired": 4, "healthy": 3, "skipped": 2, "failed": 1}, + "status_matches_object_oracle": True, + "terminal_retention_cases": list(SCANNER_HEAL_RELEASE_G05_TERMINAL_RETENTION_CASES), + "terminal_retention_window_seconds": 3600, + "max_terminal_record_age_seconds": 3599, + "terminal_records_pruned_after_window": 2, + }) + status_compat = root / "status-compat.json" + write_json(status_compat, { + "schema": 1, + "evidence_type": "measured", + "source_revision": source_revision, + "concurrent_status_cases": list(SCANNER_HEAL_RELEASE_G06_CONCURRENT_STATUS_CASES), + "status_samples": 4, + "all_status_responses_http_success": True, + "partial_status_reports_degraded": True, + "legacy_client_cases": list(SCANNER_HEAL_RELEASE_G06_LEGACY_CLIENT_CASES), + "rustfs_and_minio_paths_compatible": True, + "empty_body_status_requests_accepted": True, + "truncation_cases": list(SCANNER_HEAL_RELEASE_G06_TRUNCATION_CASES), + "truncated_payloads_rejected": True, + "max_status_payload_bytes": 4096, + }) + disposition = root / "disposition.json" + write_json(disposition, { + "schema": 1, + "evidence_type": "measured", + "source_revision": source_revision, + "manager_disposition_cases": list(SCANNER_HEAL_RELEASE_RD_MANAGER_CASES), + "manager_dispositions_are_terminal": True, + "event_disposition_cases": list(SCANNER_HEAL_RELEASE_RD_EVENT_CASES), + "events_correlate_to_manager_dispositions": True, + "ledger_disposition_cases": list(SCANNER_HEAL_RELEASE_RD_LEDGER_CASES), + "ledger_correlates_to_events": True, + "ledger_replay_preserves_terminal_disposition": True, + "grace_cases": list(SCANNER_HEAL_RELEASE_RD_GRACE_CASES), + "grace_window_seconds": 300, + "grace_retention_observed": True, + "grace_expiry_pruned_terminal_records": True, + }) + return status_outcome, status_compat, disposition + + +def run_self_test() -> None: + import tempfile + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + source_revision = git_head() + status_outcome, status_compat, disposition = write_self_test_inputs(root, source_revision) + descriptor = build_descriptor(parse_args([ + "--status-outcome-json", str(status_outcome), + "--status-compat-json", str(status_compat), + "--disposition-json", str(disposition), + "--out-dir", str(root / "out"), + "--duration-seconds", "60", + ])) + require(descriptor.is_file(), "self-test descriptor missing") + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + source_revision = git_head() + status_outcome, status_compat, disposition = write_self_test_inputs(root, source_revision) + payload = read_json(status_compat) + payload["truncation_cases"].remove("truncated-node-status-reject") + write_json(status_compat, payload) + try: + build_descriptor(parse_args([ + "--status-outcome-json", str(status_outcome), + "--status-compat-json", str(status_compat), + "--disposition-json", str(disposition), + "--out-dir", str(root / "out"), + ])) + except ValueError as err: + require("truncation_cases missing cases" in str(err), "wrong self-test failure for missing truncation") + else: + raise ValueError("self-test accepted incomplete truncation evidence") + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--status-outcome-json", type=Path) + parser.add_argument("--status-compat-json", type=Path) + parser.add_argument("--disposition-json", type=Path) + parser.add_argument("--out-dir", type=Path) + parser.add_argument("--source-revision") + parser.add_argument("--run-id") + parser.add_argument("--measurement-window-id") + parser.add_argument("--started-at") + parser.add_argument("--finished-at") + parser.add_argument("--duration-seconds", type=int, default=60) + parser.add_argument("--self-test", action="store_true") + args = parser.parse_args(argv) + if not args.self_test: + if args.status_outcome_json is None: + parser.error("--status-outcome-json is required unless --self-test is used") + if args.status_compat_json is None: + parser.error("--status-compat-json is required unless --self-test is used") + if args.disposition_json is None: + parser.error("--disposition-json is required unless --self-test is used") + if args.out_dir is None: + parser.error("--out-dir is required unless --self-test is used") + return args + + +def main() -> int: + try: + args = parse_args() + if args.self_test: + run_self_test() + return 0 + descriptor = build_descriptor(args) + print(f"Status-and-outcome release descriptor verified: {descriptor}") + return 0 + except (ValueError, KeyError, OSError, subprocess.SubprocessError) as err: + print(f"ERROR: {err}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/test_scanner_heal_checkpoint_crash_evidence.sh b/scripts/test_scanner_heal_checkpoint_crash_evidence.sh new file mode 100755 index 000000000..fe5ea7282 --- /dev/null +++ b/scripts/test_scanner_heal_checkpoint_crash_evidence.sh @@ -0,0 +1,8 @@ +#!/usr/bin/env bash + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +RUNNER="$SCRIPT_DIR/run_scanner_heal_checkpoint_crash_evidence.py" + +"${RUSTFS_PYTHON_BIN:-python3}" "$RUNNER" --self-test diff --git a/scripts/test_scanner_heal_maintenance_evidence.sh b/scripts/test_scanner_heal_maintenance_evidence.sh new file mode 100755 index 000000000..c1feecccd --- /dev/null +++ b/scripts/test_scanner_heal_maintenance_evidence.sh @@ -0,0 +1,8 @@ +#!/usr/bin/env bash + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +RUNNER="$SCRIPT_DIR/run_scanner_heal_maintenance_evidence.py" + +"${RUSTFS_PYTHON_BIN:-python3}" "$RUNNER" --self-test diff --git a/scripts/test_scanner_heal_status_outcome_evidence.sh b/scripts/test_scanner_heal_status_outcome_evidence.sh new file mode 100755 index 000000000..2df5a7fab --- /dev/null +++ b/scripts/test_scanner_heal_status_outcome_evidence.sh @@ -0,0 +1,7 @@ +#!/usr/bin/env bash +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +RUNNER="$SCRIPT_DIR/run_scanner_heal_status_outcome_evidence.py" + +"${RUSTFS_PYTHON_BIN:-python3}" "$RUNNER" --self-test