| 252 | use crate::control::array_sync::outbound::subscriber_state::SubscriberStore; |
| 253 | |
| 254 | fn make_server() -> ( |
| 255 | OriginCatchupServer, |
| 256 | Arc<ArrayDeliveryRegistry>, |
| 257 | Arc<SubscriberMap>, |
| 258 | ) { |
| 259 | let op_log = Arc::new(OriginOpLog::open_in_memory().unwrap()); |
| 260 | let snapshots = OriginSnapshotStore::open_in_memory().unwrap(); |
| 261 | let delivery = Arc::new(ArrayDeliveryRegistry::new()); |
| 262 | let store = SubscriberStore::in_memory().unwrap(); |
| 263 | let cursors = Arc::new(SubscriberMap::new(store)); |
| 264 | let ack_registry = ArrayAckRegistry::open_in_memory().unwrap(); |
| 265 | |
| 266 | // Minimal schema registry — needed only to pass the array-exists check. |
| 267 | let schema_db = Arc::new( |
| 268 | redb::Database::builder() |
| 269 | .create_with_backend(redb::backends::InMemoryBackend::new()) |
| 270 | .unwrap(), |
| 271 | ); |
| 272 | { |
| 273 | let txn = schema_db.begin_write().unwrap(); |
| 274 | txn.open_table(redb::TableDefinition::<&[u8], &[u8]>::new( |
| 275 | "array_schema_docs", |
| 276 | )) |
| 277 | .unwrap(); |
| 278 | txn.commit().unwrap(); |
| 279 | } |
| 280 | let replica_id = ReplicaId::new(0); |
| 281 | let hlc_gen = Arc::new(nodedb_array::sync::HlcGenerator::new(replica_id)); |
| 282 | let schemas = Arc::new( |
| 283 | crate::control::array_sync::OriginSchemaRegistry::open(schema_db, replica_id, hlc_gen) |
| 284 | .unwrap(), |
| 285 | ); |
| 286 | |
| 287 | let server = OriginCatchupServer::new( |
| 288 | Arc::clone(&op_log), |
| 289 | Arc::clone(&schemas), |
| 290 | snapshots, |
| 291 | Arc::clone(&delivery), |
| 292 | Arc::clone(&cursors), |
| 293 | ack_registry, |
| 294 | ); |
| 295 | |
| 296 | // Register a schema for "arr" so schema_hlc check passes. |
| 297 | // schema_registry's import_snapshot or register will be used; here |
| 298 | // we use the internal map — schemas.schema_hlc("arr") returns None |
| 299 | // unless we register. For tests, just test that unknown arrays are |
| 300 | // gracefully ignored and known ones stream ops. |
| 301 | // We skip schema registration and verify op-stream path doesn't crash. |
| 302 | (server, delivery, cursors) |
| 303 | } |
| 304 | |
| 305 | #[test] |
| 306 | fn unknown_array_is_ignored() { |