authored_draft_row_tests.rs (8917B)
1 use super::*; 2 use crate::authored_draft::{decode_row, query_tests::draft, tests::open_store}; 3 use radroots_storage::{ 4 authored_draft::{AuthoredDraft, AuthoredDraftStore}, 5 authored_draft_query::{AuthoredDraftQuery, AuthoredDraftQueryRecord}, 6 }; 7 use sqlx::{Row, ValueRef}; 8 use tempfile::TempDir; 9 10 // Fixture-only corruption, on one connection, with the exact migration guard 11 // restored before the owning reads run. No production schema is changed. 12 async fn corrupt_field(store: &crate::SqliteStorage, assignment: &'static str) { 13 let guard: String = sqlx::query_scalar( 14 "SELECT sql FROM sqlite_schema WHERE name = 'radroots_runtime_authored_draft_revisions_update_guard'", 15 ).fetch_one(store.pool()).await.unwrap(); 16 let mut connection = store.pool().acquire().await.unwrap(); 17 sqlx::query("DROP TRIGGER radroots_runtime_authored_draft_revisions_update_guard") 18 .execute(&mut *connection) 19 .await 20 .unwrap(); 21 sqlx::query("PRAGMA ignore_check_constraints = ON") 22 .execute(&mut *connection) 23 .await 24 .unwrap(); 25 // Only closed test literals below supply this assignment. 26 let statement = format!("UPDATE radroots_runtime_authored_draft_revisions SET {assignment}"); 27 sqlx::query(sqlx::AssertSqlSafe(statement.as_str())) 28 .execute(&mut *connection) 29 .await 30 .unwrap(); 31 sqlx::query("PRAGMA ignore_check_constraints = OFF") 32 .execute(&mut *connection) 33 .await 34 .unwrap(); 35 sqlx::query(sqlx::AssertSqlSafe(guard.as_str())) 36 .execute(&mut *connection) 37 .await 38 .unwrap(); 39 } 40 41 async fn rejects_point_reads_and_replay(store: &crate::SqliteStorage, value: &AuthoredDraft) { 42 assert_eq!( 43 store.authored_draft_head(value.draft_id()).await, 44 Err(Error::CorruptAuthoredDraft) 45 ); 46 assert_eq!( 47 store 48 .authored_draft_revision(value.draft_id(), value.revision()) 49 .await, 50 Err(Error::CorruptAuthoredDraft) 51 ); 52 assert_eq!( 53 store.append_authored_draft(value.clone(), None).await, 54 Err(Error::CorruptAuthoredDraft) 55 ); 56 let successor = value.successor(vec![2], value.stage(), None, 11).unwrap(); 57 assert_eq!( 58 store 59 .append_authored_draft(successor, Some(value.revision())) 60 .await, 61 Err(Error::CorruptAuthoredDraft) 62 ); 63 let mut transaction = store.pool().begin().await.unwrap(); 64 assert_eq!( 65 crate::authored_draft::load_head_tx(&mut transaction, value.draft_id()).await, 66 Err(Error::CorruptAuthoredDraft) 67 ); 68 transaction.rollback().await.unwrap(); 69 } 70 71 #[tokio::test] 72 async fn oversized_columns_are_bounded_in_sql_and_remain_corrupt() { 73 for (assignment, column, sentinel) in [ 74 ("snapshot = zeroblob(16777217)", "snapshot", false), 75 ("author = zeroblob(1048576)", "author", false), 76 ( 77 "payload_sha256 = zeroblob(1048576)", 78 "payload_sha256", 79 false, 80 ), 81 ("operation_id = zeroblob(1048576)", "operation_id", true), 82 ("payload_scope = zeroblob(1048576)", "payload_scope", true), 83 ( 84 "payload_schema = replace(hex(zeroblob(65536)), '0', 'é')", 85 "payload_schema", 86 false, 87 ), 88 ] { 89 let temp = TempDir::new().unwrap(); 90 let store = open_store(&temp).await; 91 let value = draft(1, "known.v1", None, vec![1]); 92 store 93 .append_authored_draft(value.clone(), None) 94 .await 95 .unwrap(); 96 corrupt_field(&store, assignment).await; 97 let row = load( 98 store.pool(), 99 value.draft_id().as_bytes(), 100 Some(value.revision()), 101 ) 102 .await 103 .unwrap() 104 .unwrap(); 105 if sentinel { 106 assert_eq!( 107 row.try_get::<Option<Vec<u8>>, _>(column).unwrap(), 108 Some(vec![0]) 109 ); 110 } else { 111 assert!(row.try_get_raw(column).unwrap().is_null(), "{column}"); 112 } 113 assert_eq!(decode_row(&row), Err(Error::CorruptAuthoredDraft)); 114 rejects_point_reads_and_replay(&store, &value).await; 115 // Foreign/corrupt authors remain outside this independently supplied 116 // author selection; every other malformed head remains a locator. 117 if column != "author" { 118 let page = store 119 .query_authored_drafts( 120 AuthoredDraftQuery::for_author_all_schemas([7; 32], 1).unwrap(), 121 ) 122 .await 123 .unwrap(); 124 assert!(matches!( 125 page.records(), 126 [AuthoredDraftQueryRecord::Corrupt { .. }] 127 )); 128 } 129 store.close().await.unwrap(); 130 } 131 } 132 133 #[tokio::test] 134 async fn snapshot_limit_is_inclusive_and_empty_snapshot_is_rejected_before_decode() { 135 for (assignment, expected) in [ 136 ("snapshot = zeroblob(16777216)", Some(16777216)), 137 ("snapshot = X''", None), 138 ] { 139 let temp = TempDir::new().unwrap(); 140 let store = open_store(&temp).await; 141 let value = draft(1, "known.v1", None, vec![1]); 142 store 143 .append_authored_draft(value.clone(), None) 144 .await 145 .unwrap(); 146 corrupt_field(&store, assignment).await; 147 let row = load(store.pool(), value.draft_id().as_bytes(), None) 148 .await 149 .unwrap() 150 .unwrap(); 151 assert_eq!( 152 row.try_get::<Option<Vec<u8>>, _>("snapshot") 153 .unwrap() 154 .map(|bytes| bytes.len()), 155 expected 156 ); 157 assert_eq!(decode_row(&row), Err(Error::CorruptAuthoredDraft)); 158 store.close().await.unwrap(); 159 } 160 } 161 162 #[tokio::test] 163 async fn malformed_key_fails_inventory_instead_of_becoming_absence() { 164 let temp = TempDir::new().unwrap(); 165 let store = open_store(&temp).await; 166 store 167 .append_authored_draft(draft(1, "known.v1", None, vec![1]), None) 168 .await 169 .unwrap(); 170 corrupt_field(&store, "draft_id = zeroblob(1048576)").await; 171 assert!( 172 store 173 .query_authored_drafts(AuthoredDraftQuery::for_author_all_schemas([7; 32], 1).unwrap()) 174 .await 175 .is_err() 176 ); 177 store.close().await.unwrap(); 178 } 179 180 #[tokio::test] 181 async fn bounded_load_preserves_history_absence_and_compound_index() { 182 let temp = TempDir::new().unwrap(); 183 let store = open_store(&temp).await; 184 let first = draft(1, "known.v1", None, vec![1]); 185 let second = first.successor(vec![2], first.stage(), None, 11).unwrap(); 186 store 187 .append_authored_draft(first.clone(), None) 188 .await 189 .unwrap(); 190 store 191 .append_authored_draft(second.clone(), Some(first.revision())) 192 .await 193 .unwrap(); 194 for (revision, expected) in [(None, &second), (Some(first.revision()), &first)] { 195 let row = load(store.pool(), first.draft_id().as_bytes(), revision) 196 .await 197 .unwrap() 198 .unwrap(); 199 assert_eq!(&decode_row(&row).unwrap(), expected); 200 assert!( 201 row.try_get::<Option<Vec<u8>>, _>("operation_id") 202 .unwrap() 203 .is_none() 204 ); 205 assert!( 206 row.try_get::<Option<Vec<u8>>, _>("payload_scope") 207 .unwrap() 208 .is_none() 209 ); 210 } 211 assert!(load(store.pool(), &[2; 16], None).await.unwrap().is_none()); 212 assert!( 213 load( 214 store.pool(), 215 first.draft_id().as_bytes(), 216 Some(AuthoredDraftRevision::new(3).unwrap()) 217 ) 218 .await 219 .unwrap() 220 .is_none() 221 ); 222 assert!(matches!( 223 load( 224 store.pool(), 225 &[1; 16], 226 Some(AuthoredDraftRevision::new(u64::MAX).unwrap()) 227 ) 228 .await, 229 Err(Error::InvalidAuthoredDraft) 230 )); 231 for statement in [HEAD, REVISION] { 232 let explain = format!("EXPLAIN QUERY PLAN {statement}"); 233 let rows = sqlx::query(sqlx::AssertSqlSafe(explain.as_str())) 234 .bind([1_u8; 16].as_slice()) 235 .bind(1_i64) 236 .fetch_all(store.pool()) 237 .await 238 .unwrap(); 239 let details: Vec<String> = rows.iter().map(|row| row.get("detail")).collect(); 240 assert!( 241 details.iter().any(|detail| detail.contains("SEARCH") 242 && detail.contains("PRIMARY KEY") 243 && detail.contains("draft_id=?")), 244 "{details:?}" 245 ); 246 assert!( 247 !details 248 .iter() 249 .any(|detail| detail.contains("SCAN ") || detail.contains("TEMP B-TREE")), 250 "{details:?}" 251 ); 252 } 253 store.close().await.unwrap(); 254 assert!(matches!( 255 load(store.pool(), &[1; 16], None).await, 256 Err(Error::BackendUnavailable) 257 )); 258 }