@@ -232,6 +232,7 @@ struct ImportSessionGuard {
232232 executor : JsCallbackExecutor ,
233233 pending : Option < PendingImport > ,
234234 serialize : Option < OwnedMutexGuard < ( ) > > ,
235+ ownership : Option < super :: pipeline:: runner:: leadership:: LeadershipGuard > ,
235236}
236237
237238enum PendingImport {
@@ -275,9 +276,11 @@ impl Drop for ImportSessionGuard {
275276 } ;
276277 let executor = self . executor . clone ( ) ;
277278 let serialize = self . serialize . take ( ) ;
279+ let ownership = self . ownership . take ( ) ;
278280 wasm_bindgen_utils:: prelude:: wasm_bindgen_futures:: spawn_local ( async move {
279281 // Keep later queries/imports serialized until rollback finishes.
280282 let _serialize = serialize;
283+ let _ownership = ownership;
281284 let id = match pending {
282285 PendingImport :: Active { id, in_flight } => {
283286 // The SDK rejects cancellation while an append/finish/cancel
@@ -410,6 +413,29 @@ impl JsCallbackExecutor {
410413 } )
411414 } )
412415 . collect :: < Result < Vec < _ > , _ > > ( ) ?;
416+ // Both cohort and per-network imports hold the same cross-tab lock
417+ // before reading watermarks. A competing import cannot populate the DB
418+ // between this preflight and opening our worker session.
419+ let started = js_sys:: Date :: now ( ) ;
420+ let ownership = loop {
421+ match super :: pipeline:: runner:: leadership:: acquire_import ( ) . await {
422+ Ok ( Some ( guard) ) => break Some ( guard) ,
423+ Ok ( None ) if within_import_wait ( started) => {
424+ gloo_timers:: future:: TimeoutFuture :: new ( IMPORT_WAIT_POLL_MS ) . await ;
425+ }
426+ Ok ( None ) => {
427+ return Err ( LocalDbQueryError :: database (
428+ "Timed out waiting for import ownership" ,
429+ ) ) ;
430+ }
431+ Err ( error) => {
432+ // Import still works through the SDK lease without Web
433+ // Locks, but preflight cannot safely justify index drops.
434+ tracing:: warn!( %error, "Import ownership unavailable; retaining indexes" ) ;
435+ break None ;
436+ }
437+ }
438+ } ;
413439 // Only a globally fresh DB benefits from rebuilding every index. Adding
414440 // seeds to a populated DB keeps indexes over its existing history intact.
415441 let watermarks = self
@@ -419,7 +445,7 @@ impl JsCallbackExecutor {
419445 . await ?;
420446 let has_watermarks: Vec < serde_json:: Value > =
421447 serde_json:: from_str ( & watermarks) . map_err ( |_| LocalDbQueryError :: invalid_response ( ) ) ?;
422- let defer_indexes = has_watermarks. is_empty ( ) ;
448+ let defer_indexes = ownership . is_some ( ) && has_watermarks. is_empty ( ) ;
423449 let mut prefix = String :: from ( "BEGIN;\n " ) ;
424450 let mut suffix = String :: new ( ) ;
425451 if defer_indexes {
@@ -445,6 +471,7 @@ impl JsCallbackExecutor {
445471 executor : self . clone ( ) ,
446472 pending : None ,
447473 serialize : Some ( serialize) ,
474+ ownership,
448475 } ;
449476 let started = js_sys:: Date :: now ( ) ;
450477 let id = loop {
@@ -457,6 +484,11 @@ impl JsCallbackExecutor {
457484 match result {
458485 Err ( err) if err. is_import_in_progress ( ) && within_import_wait ( started) => {
459486 guard. pending = None ;
487+ // An older SDK caller may not participate in our Web Lock.
488+ // Its import invalidates preflight; retain indexes rather
489+ // than rebuild over data committed while we waited.
490+ prefix = String :: from ( "BEGIN;\n " ) ;
491+ suffix = String :: from ( "COMMIT;\n " ) ;
460492 gloo_timers:: future:: TimeoutFuture :: new ( IMPORT_WAIT_POLL_MS ) . await ;
461493 }
462494 result => break result?,
@@ -1318,6 +1350,41 @@ pub mod tests {
13181350 }
13191351 }
13201352
1353+ #[ wasm_bindgen_test]
1354+ async fn import_lease_contention_retains_preflight_indexes ( ) {
1355+ let local_db = Function :: new_no_args ( r#"return {
1356+ chunks: [], begins: 0,
1357+ query(sql) { return {value: sql.includes('target_watermarks') ? '[]' :
1358+ '[{"name":"idx_t_value","sql":"CREATE INDEX idx_t_value ON t(value)"}]'}; },
1359+ transaction() {}, wipeAndRecreate() {},
1360+ beginSqlDumpImport() {
1361+ if (++this.begins === 1) return {error:{msg:'busy',readableMsg:'SQL dump import is in progress'}};
1362+ return {value:'session'};
1363+ },
1364+ appendSqlDumpChunk(id, sql) { this.chunks.push(sql); return {value:''}; },
1365+ finishSqlDumpImport() { return {value:''}; }, cancelSqlDumpImport() {}
1366+ };"# ) . call0 ( & JsValue :: UNDEFINED ) . unwrap ( ) ;
1367+ let exec = JsCallbackExecutor :: new ( local_db. clone ( ) ) . unwrap ( ) ;
1368+ exec. execute_sql_dump ( Arc :: new ( "BEGIN; INSERT INTO t VALUES (1); COMMIT;" . into ( ) ) )
1369+ . await
1370+ . unwrap ( ) ;
1371+ let chunks = Reflect :: get ( & local_db, & "chunks" . into ( ) )
1372+ . unwrap ( )
1373+ . dyn_into :: < Array > ( )
1374+ . unwrap ( ) ;
1375+ let sql = chunks
1376+ . iter ( )
1377+ . map ( |v| v. as_string ( ) . unwrap ( ) )
1378+ . collect :: < String > ( ) ;
1379+ assert ! ( sql. contains( "INSERT INTO t VALUES (1)" ) ) ;
1380+ assert ! ( !sql. contains( "DROP INDEX" ) ) ;
1381+ assert ! ( !sql. contains( "CREATE INDEX" ) ) ;
1382+ assert_eq ! (
1383+ Reflect :: get( & local_db, & "begins" . into( ) ) . unwrap( ) . as_f64( ) ,
1384+ Some ( 2.0 )
1385+ ) ;
1386+ }
1387+
13211388 #[ wasm_bindgen_test]
13221389 async fn cohort_rejects_bad_envelopes_before_opening_a_session ( ) {
13231390 let local_db = Function :: new_no_args (
0 commit comments