@@ -56,14 +56,6 @@ pub(super) const AUDIT_PAGE_ROWS: i64 = 128;
5656/// limit while avoiding tens of thousands of SQL-channel round trips on a
5757/// production-sized store.
5858pub ( super ) const OBSERVATION_AUDIT_PAGE_ROWS : i64 = 48 ;
59- const SESSION_TEMPORAL_REPAIR_AUDITS : & [ & str ] = & [
60- "session temporal receipts or cursor keys are mutable" ,
61- "session cursor key rotation state is invalid" ,
62- "session refresh operation state is invalid" ,
63- "session temporal generation state is invalid" ,
64- "session temporal authority ownership is invalid" ,
65- ] ;
66- pub const SESSION_TEMPORAL_REPAIR_AUDIT_PAGE_ROWS : i64 = 256 ;
6759
6860pub async fn authority_invariant_triggers_intact (
6961 conn : & impl QueryExecutor ,
@@ -527,222 +519,16 @@ pub async fn validate_authority_rows_exhaustive(
527519 validate_invariant_rows ( conn) . await
528520}
529521
530- pub async fn validate_session_temporal_repair_authority_audit (
531- conn : & impl QueryExecutor ,
532- audit_index : usize ,
533- ) -> tracedecay_runtime_core:: errors:: Result < ( ) > {
534- let invariant = INVARIANTS
535- . iter ( )
536- . filter ( |invariant| SESSION_TEMPORAL_REPAIR_AUDITS . contains ( & invariant. violation ) )
537- . nth ( audit_index)
538- . ok_or_else ( || {
539- global_db_operation_message (
540- OPERATION ,
541- format ! ( "unknown session temporal repair authority audit {audit_index}" ) ,
542- )
543- } ) ?;
544- if let Some ( query) = invariant. audit_query
545- && query_has_rows ( conn, query) . await ?
546- {
547- return Err ( global_db_operation_message ( OPERATION , invariant. violation ) ) ;
548- }
549- Ok ( ( ) )
550- }
551-
552- #[ cfg( test) ]
553- pub async fn validate_session_temporal_effect_authority_page (
554- conn : & impl QueryExecutor ,
555- after_rowid : i64 ,
556- ) -> tracedecay_runtime_core:: errors:: Result < ( i64 , bool ) > {
557- validate_session_temporal_effect_authority_page_with_limit (
558- conn,
559- after_rowid,
560- SESSION_TEMPORAL_REPAIR_AUDIT_PAGE_ROWS ,
561- )
562- . await
563- }
564-
565- pub async fn validate_session_temporal_effect_authority_page_with_limit (
566- conn : & impl QueryExecutor ,
567- after_rowid : i64 ,
568- page_rows : i64 ,
569- ) -> tracedecay_runtime_core:: errors:: Result < ( i64 , bool ) > {
570- debug_assert ! ( page_rows > 0 ) ;
571- let mut rows = conn
572- . query (
573- "SELECT effect.rowid, observation.observation_id IS NULL
574- FROM session_temporal_observation_effects AS effect
575- LEFT JOIN observations AS observation
576- ON observation.observation_id = effect.observation_id
577- AND observation.sequence = effect.observation_sequence
578- AND observation.receipt_id = effect.receipt_id
579- WHERE effect.rowid > ?1
580- ORDER BY effect.rowid
581- LIMIT ?2" ,
582- params ! [ after_rowid, page_rows] ,
583- )
584- . await
585- . map_err ( |error| global_db_operation_error ( OPERATION , error) ) ?;
586- let mut last_rowid = after_rowid;
587- let mut count = 0_i64 ;
588- while let Some ( row) = rows
589- . next ( )
590- . await
591- . map_err ( |error| global_db_operation_error ( OPERATION , error) ) ?
592- {
593- last_rowid = row
594- . get ( 0 )
595- . map_err ( |error| global_db_operation_error ( OPERATION , error) ) ?;
596- let invalid = row
597- . get :: < i64 > ( 1 )
598- . map_err ( |error| global_db_operation_error ( OPERATION , error) ) ?
599- != 0 ;
600- if invalid {
601- return Err ( global_db_operation_message (
602- OPERATION ,
603- SESSION_TEMPORAL_REPAIR_AUDITS [ 0 ] ,
604- ) ) ;
605- }
606- count += 1 ;
607- }
608- Ok ( ( last_rowid, count < page_rows) )
609- }
610-
611- pub async fn validate_session_temporal_receipt_authority_page_with_limit (
612- conn : & impl QueryExecutor ,
613- after_rowid : i64 ,
614- page_rows : i64 ,
615- ) -> tracedecay_runtime_core:: errors:: Result < ( i64 , bool ) > {
616- debug_assert ! ( page_rows > 0 ) ;
617- let mut rows = conn
618- . query (
619- "SELECT receipt.rowid,
620- generation.session_id IS NULL
621- OR (
622- receipt.batch_ordinal > 0
623- AND NOT EXISTS (
624- SELECT 1
625- FROM session_temporal_projection_receipts AS previous
626- WHERE previous.session_id = receipt.session_id
627- AND previous.generation = receipt.generation
628- AND previous.batch_ordinal = receipt.batch_ordinal - 1
629- AND previous.source_through <= receipt.source_through
630- AND previous.projection_through <= receipt.projection_through
631- )
632- )
633- FROM session_temporal_projection_receipts AS receipt
634- LEFT JOIN session_temporal_generations AS generation
635- ON generation.session_id = receipt.session_id
636- AND generation.generation = receipt.generation
637- AND generation.frozen_watermarks_json = receipt.frozen_watermarks_json
638- WHERE receipt.rowid > ?1
639- ORDER BY receipt.rowid
640- LIMIT ?2" ,
641- params ! [ after_rowid, page_rows] ,
642- )
643- . await
644- . map_err ( |error| global_db_operation_error ( OPERATION , error) ) ?;
645- let mut last_rowid = after_rowid;
646- let mut count = 0_i64 ;
647- while let Some ( row) = rows
648- . next ( )
649- . await
650- . map_err ( |error| global_db_operation_error ( OPERATION , error) ) ?
651- {
652- last_rowid = row
653- . get ( 0 )
654- . map_err ( |error| global_db_operation_error ( OPERATION , error) ) ?;
655- let invalid = row
656- . get :: < i64 > ( 1 )
657- . map_err ( |error| global_db_operation_error ( OPERATION , error) ) ?
658- != 0 ;
659- if invalid {
660- return Err ( global_db_operation_message (
661- OPERATION ,
662- SESSION_TEMPORAL_REPAIR_AUDITS [ 0 ] ,
663- ) ) ;
664- }
665- count += 1 ;
666- }
667- Ok ( ( last_rowid, count < page_rows) )
668- }
669-
670522#[ cfg( test) ]
671523mod tests {
672524 use tempfile:: TempDir ;
673525
674526 use super :: {
675527 FOREIGN_KEY_AUDIT_PROGRESS , foreign_key_violation_exists_read_only,
676- foreign_key_violation_exists_resumable, validate_session_temporal_effect_authority_page,
677- validate_session_temporal_effect_authority_page_with_limit,
528+ foreign_key_violation_exists_resumable,
678529 } ;
679530 use tracedecay_runtime_core:: db:: engine:: TestConnection ;
680531
681- #[ tokio:: test]
682- async fn session_temporal_effect_audit_checkpoints_bounded_pages ( ) {
683- let directory = TempDir :: new ( ) . unwrap ( ) ;
684- let database_path = directory. path ( ) . join ( "sessions.db" ) ;
685- let mut connection = rusqlite:: Connection :: open ( & database_path) . unwrap ( ) ;
686- connection
687- . execute_batch (
688- "CREATE TABLE observations (
689- observation_id TEXT NOT NULL,
690- sequence INTEGER NOT NULL,
691- receipt_id TEXT NOT NULL,
692- PRIMARY KEY(observation_id, sequence, receipt_id)
693- );
694- CREATE TABLE session_temporal_observation_effects (
695- observation_id TEXT NOT NULL,
696- observation_sequence INTEGER NOT NULL,
697- receipt_id TEXT NOT NULL
698- );" ,
699- )
700- . unwrap ( ) ;
701- let transaction = connection. transaction ( ) . unwrap ( ) ;
702- for ordinal in 1 ..=257 {
703- let observation_id = format ! ( "observation-{ordinal}" ) ;
704- let receipt_id = format ! ( "receipt-{ordinal}" ) ;
705- transaction
706- . execute (
707- "INSERT INTO observations(observation_id, sequence, receipt_id)
708- VALUES (?1, ?2, ?3)" ,
709- rusqlite:: params![ observation_id, ordinal, receipt_id] ,
710- )
711- . unwrap ( ) ;
712- transaction
713- . execute (
714- "INSERT INTO session_temporal_observation_effects(
715- observation_id, observation_sequence, receipt_id
716- ) VALUES (?1, ?2, ?3)" ,
717- rusqlite:: params![ observation_id, ordinal, receipt_id] ,
718- )
719- . unwrap ( ) ;
720- }
721- transaction. commit ( ) . unwrap ( ) ;
722- drop ( connection) ;
723- let connection = TestConnection :: open ( & database_path) ;
724-
725- let ( first_cursor, first_complete) =
726- validate_session_temporal_effect_authority_page ( & connection, 0 )
727- . await
728- . unwrap ( ) ;
729- assert_eq ! ( first_cursor, 256 ) ;
730- assert ! ( !first_complete) ;
731- let ( second_cursor, second_complete) =
732- validate_session_temporal_effect_authority_page ( & connection, first_cursor)
733- . await
734- . unwrap ( ) ;
735- assert_eq ! ( second_cursor, 257 ) ;
736- assert ! ( second_complete) ;
737-
738- let ( adaptive_cursor, adaptive_complete) =
739- validate_session_temporal_effect_authority_page_with_limit ( & connection, 0 , 512 )
740- . await
741- . unwrap ( ) ;
742- assert_eq ! ( adaptive_cursor, 257 ) ;
743- assert ! ( adaptive_complete) ;
744- }
745-
746532 #[ tokio:: test]
747533 async fn foreign_key_audit_finds_violations_by_child_table ( ) {
748534 let directory = TempDir :: new ( ) . unwrap ( ) ;
0 commit comments