+768

.agent/logs/20260203_collection_backfill_race_conditioon.md

View file Reviewed

··· 1 1 + # Collection Backfill Race Condition Analysis 2 2 + 3 3 + **Date:** 2026-02-03 4 4 + **Issue:** Lost updates during concurrent collection item processing 5 5 + **Status:** Root cause identified 6 6 + 7 7 + --- 8 8 + 9 9 + ## Executive Summary 10 10 + 11 11 + During account data sync, when multiple collection items are added to the same collection concurrently, only the last item appears in the final result. This is caused by a **classic lost update race condition** in the collection repository's save method, which uses a DELETE-all-then-INSERT-all pattern that causes concurrent transactions to overwrite each other's changes. 12 12 + 13 13 + **Expected:** Collection A has 3 items, Collection B has 2 items 14 14 + **Actual:** Collection A has 1 item, Collection B has 1 item 15 15 + **Data Lost:** 3 out of 5 collection items 16 16 + 17 17 + --- 18 18 + 19 19 + ## Problem Symptoms 20 20 + 21 21 + From production logs, we can see: 22 22 + 23 23 + ``` 24 24 + [SYNC] [SYNC] Retrieved 5 records from at.margin.collectionItem (page 1) 25 25 + [SYNC] [SYNC] Processing batch 1 (5 records) from page 1... 26 26 + 27 27 + # All 5 items resolve successfully 28 28 + [SYNC] [FirehoseWorker] Resolved Margin collection item - collectionId: 63d0727a..., cardId: 7c7ae767..., itemUri: ...3mdlhi4wewh2h 29 29 + [SYNC] [FirehoseWorker] Resolved Margin collection item - collectionId: ce2891f7..., cardId: eca2df55..., itemUri: ...3mdykmh3pv22x 30 30 + [SYNC] [FirehoseWorker] Resolved Margin collection item - collectionId: ce2891f7..., cardId: 0a23c9b0..., itemUri: ...3mdyknieydy2a 31 31 + [SYNC] [FirehoseWorker] Resolved Margin collection item - collectionId: 63d0727a..., cardId: e8f84f9f..., itemUri: ...3mdvzqmvl6o2x 32 32 + [SYNC] [FirehoseWorker] Resolved Margin collection item - collectionId: 63d0727a..., cardId: 9fd4381a..., itemUri: ...3mdykgokcyp2m 33 33 + 34 34 + # All 5 successfully added 35 35 + [SYNC] [FirehoseWorker] Successfully added Margin annotation to collection - user: did:plc:rlknsba2qldjkicxsmni3vyn, cardId: 7c7ae767..., collectionId: 63d0727a... 36 36 + [SYNC] [FirehoseWorker] Successfully added Margin annotation to collection - user: did:plc:rlknsba2qldjkicxsmni3vyn, cardId: eca2df55..., collectionId: ce2891f7... 37 37 + [SYNC] [FirehoseWorker] Successfully added Margin annotation to collection - user: did:plc:rlknsba2qldjkicxsmni3vyn, cardId: e8f84f9f..., collectionId: 63d0727a... 38 38 + [SYNC] [FirehoseWorker] Successfully added Margin annotation to collection - user: did:plc:rlknsba2qldjkicxsmni3vyn, cardId: 0a23c9b0..., collectionId: ce2891f7... 39 39 + [SYNC] [FirehoseWorker] Successfully added Margin annotation to collection - user: did:plc:rlknsba2qldjkicxsmni3vyn, cardId: 9fd4381a..., collectionId: 63d0727a... 40 40 + 41 41 + [SYNC] [SYNC] Batch 1: processed 5/5 records 42 42 + [SYNC] [SYNC] Processed 5 collection items 43 43 + ``` 44 44 + 45 45 + **Expected Distribution:** 46 46 + 47 47 + - Collection `63d0727a`: 3 items (cards: 7c7ae767, e8f84f9f, 9fd4381a) 48 48 + - Collection `ce2891f7`: 2 items (cards: eca2df55, 0a23c9b0) 49 49 + 50 50 + **Actual Result:** Only 1 item in each collection (database query would show this) 51 51 + 52 52 + --- 53 53 + 54 54 + ## Root Cause: Lost Update Race Condition 55 55 + 56 56 + The issue occurs because of concurrent processing combined with a problematic database update pattern. 57 57 + 58 58 + ### Concurrent Processing 59 59 + 60 60 + **File:** `src/modules/sync/application/useCases/SyncAccountDataUseCase.ts` 61 61 + 62 62 + **Lines 231-245:** Collection items are processed in batches with concurrent execution: 63 63 + 64 64 + ```typescript 65 65 + // Process in batches of CONCURRENT_BATCH_SIZE 66 66 + for (let i = 0; i < records.length; i += CONCURRENT_BATCH_SIZE) { 67 67 + const batch = records.slice(i, i + CONCURRENT_BATCH_SIZE); 68 68 + const batchNumber = Math.floor(i / CONCURRENT_BATCH_SIZE) + 1; 69 69 + 70 70 + console.log( 71 71 + `[SYNC] Processing batch ${batchNumber} (${batch.length} records) from page ${pageCount}...`, 72 72 + ); 73 73 + 74 74 + // Process all records in this batch concurrently 75 75 + const results = await Promise.allSettled( 76 76 + batch.map((record) => 77 77 + this.processRecord(record, useCase, curatorId, collectionType), 78 78 + ), 79 79 + ); 80 80 + // ... 81 81 + } 82 82 + ``` 83 83 + 84 84 + **Key Point:** `CONCURRENT_BATCH_SIZE = 10` (line 208), so up to 10 collection items process simultaneously. 85 85 + 86 86 + --- 87 87 + 88 88 + ### The Problematic Save Pattern 89 89 + 90 90 + **File:** `src/modules/cards/infrastructure/repositories/DrizzleCollectionRepository.ts` 91 91 + 92 92 + **Lines 504-529:** The save method uses a DELETE-all-then-INSERT-all pattern: 93 93 + 94 94 + ```typescript 95 95 + async save(collection: Collection): Promise<Result<void, Error>> { 96 96 + // ... (earlier code preparing data) 97 97 + 98 98 + await this.db.transaction(async (tx) => { 99 99 + // ... (upsert collection record) 100 100 + 101 101 + // Delete existing collaborators and card links 102 102 + await tx 103 103 + .delete(collectionCollaborators) 104 104 + .where(eq(collectionCollaborators.collectionId, collectionData.id)); 105 105 + 106 106 + await tx 107 107 + .delete(collectionCards) 108 108 + .where(eq(collectionCards.collectionId, collectionData.id)); 109 109 + 110 110 + // Insert new collaborators 111 111 + if (collaborators.length > 0) { 112 112 + await tx.insert(collectionCollaborators).values(collaborators); 113 113 + } 114 114 + 115 115 + // Insert new card links with mapped published record IDs 116 116 + if (cardLinks.length > 0) { 117 117 + const cardLinksWithMappedRecords = cardLinks.map((link) => ({ 118 118 + ...link, 119 119 + publishedRecordId: link.publishedRecordId 120 120 + ? linkPublishedRecordMap.get(link.publishedRecordId) || 121 121 + link.publishedRecordId 122 122 + : undefined, 123 123 + })); 124 124 + 125 125 + await tx.insert(collectionCards).values(cardLinksWithMappedRecords); 126 126 + } 127 127 + }); 128 128 + } 129 129 + ``` 130 130 + 131 131 + **The Problem:** This pattern completely replaces all collection cards based on the in-memory collection state, regardless of what other concurrent transactions might be doing. 132 132 + 133 133 + --- 134 134 + 135 135 + ## Complete Code Flow 136 136 + 137 137 + Here's the full path from sync to database for adding a collection item: 138 138 + 139 139 + ### 1. Sync Initiation 140 140 + 141 141 + **File:** `src/modules/sync/application/useCases/SyncAccountDataUseCase.ts` 142 142 + 143 143 + **Line 184-187:** Sync collection items (third phase after bookmarks and collections): 144 144 + 145 145 + ```typescript 146 146 + // 3. Sync collection items (links cards to collections) 147 147 + console.log(`[SYNC] Syncing collection items for ${curatorId}...`); 148 148 + const itemsProcessed = await this.syncCollection( 149 149 + curatorId, 150 150 + ATPROTO_NSID.MARGIN.COLLECTION_ITEM, 151 151 + this.processMarginCollectionItemUseCase, 152 152 + ); 153 153 + ``` 154 154 + 155 155 + ### 2. Process Individual Collection Item 156 156 + 157 157 + **File:** `src/modules/atproto/application/useCases/ProcessMarginCollectionItemFirehoseEventUseCase.ts` 158 158 + 159 159 + **Lines 152-162:** Call UpdateUrlCardAssociationsUseCase: 160 160 + 161 161 + ```typescript 162 162 + const result = await this.updateUrlCardAssociationsUseCase.execute({ 163 163 + cardId: cardId.value.getStringValue(), 164 164 + curatorId: curatorDid, 165 165 + addToCollections: [collectionId.value.getStringValue()], 166 166 + context: OperationContext.FIREHOSE_EVENT, 167 167 + publishedRecordIds: { 168 168 + collectionLinks: collectionLinkMap, 169 169 + }, 170 170 + viaCardId: undefined, 171 171 + timestamp: timestamp, 172 172 + }); 173 173 + ``` 174 174 + 175 175 + ### 3. Update URL Card Associations 176 176 + 177 177 + **File:** `src/modules/cards/application/useCases/commands/UpdateUrlCardAssociationsUseCase.ts` 178 178 + 179 179 + **Lines 314-321:** Call CardCollectionService to add card to collections: 180 180 + 181 181 + ```typescript 182 182 + const addToCollectionsResult = 183 183 + await this.cardCollectionService.addCardToCollections( 184 184 + urlCard, 185 185 + collectionIds, 186 186 + curatorId, 187 187 + viaCardId, 188 188 + collectionOptions, 189 189 + ); 190 190 + ``` 191 191 + 192 192 + ### 4. Card Collection Service 193 193 + 194 194 + **File:** `src/modules/cards/domain/services/CardCollectionService.ts` 195 195 + 196 196 + **Lines 144-174:** Loop through collections and add card to each: 197 197 + 198 198 + ```typescript 199 199 + async addCardToCollections( 200 200 + card: Card, 201 201 + collectionIds: CollectionId[], 202 202 + addedBy: string, 203 203 + viaCard?: CardId, 204 204 + options?: CollectionCardOptions, 205 205 + ): Promise<Result<Collection[], Error>> { 206 206 + const updatedCollections: Collection[] = []; 207 207 + 208 208 + for (const collectionId of collectionIds) { 209 209 + const result = await this.addCardToCollection( 210 210 + collectionId, 211 211 + card, 212 212 + addedBy, 213 213 + viaCard, 214 214 + options, 215 215 + ); 216 216 + 217 217 + if (result.isErr()) { 218 218 + return err(result.error); 219 219 + } 220 220 + 221 221 + updatedCollections.push(result.value); 222 222 + } 223 223 + 224 224 + return ok(updatedCollections); 225 225 + } 226 226 + ``` 227 227 + 228 228 + ### 5. Add Card to Single Collection 229 229 + 230 230 + **File:** `src/modules/cards/domain/services/CardCollectionService.ts` 231 231 + 232 232 + **Lines 38-142:** The critical read-modify-write sequence: 233 233 + 234 234 + ```typescript 235 235 + async addCardToCollection( 236 236 + collectionId: CollectionId, 237 237 + card: Card, 238 238 + addedBy: string, 239 239 + viaCard?: CardId, 240 240 + options?: CollectionCardOptions, 241 241 + ): Promise<Result<Collection, Error>> { 242 242 + // ... validation ... 243 243 + 244 244 + // 📖 READ: Fetch collection from database 245 245 + const collectionResult = 246 246 + await this.collectionRepository.findById(collectionId); 247 247 + if (collectionResult.isErr()) { 248 248 + return err(collectionResult.error); 249 249 + } 250 250 + const collection = collectionResult.value; 251 251 + if (!collection) { 252 252 + return err(new Error(`Collection with id ${collectionId} not found`)); 253 253 + } 254 254 + 255 255 + // ... more validation ... 256 256 + 257 257 + // ✏️ MODIFY: Add card to in-memory collection 258 258 + const addCardResult = collection.addCard( 259 259 + card.id, 260 260 + addedBy, 261 261 + viaCard, 262 262 + options?.publishedRecordId, 263 263 + options?.timestamp, 264 264 + ); 265 265 + 266 266 + if (addCardResult.isErr()) { 267 267 + return err(addCardResult.error); 268 268 + } 269 269 + 270 270 + // ... increment library count ... 271 271 + 272 272 + // 💾 WRITE: Save collection back to database 273 273 + const saveCollectionResult = 274 274 + await this.collectionRepository.save(collection); 275 275 + if (saveCollectionResult.isErr()) { 276 276 + return err(saveCollectionResult.error); 277 277 + } 278 278 + 279 279 + return ok(collection); 280 280 + } 281 281 + ``` 282 282 + 283 283 + **Critical Issue:** This is a classic **Read-Modify-Write** pattern without any concurrency control! 284 284 + 285 285 + ### 6. Repository Save (The Problem!) 286 286 + 287 287 + **File:** `src/modules/cards/infrastructure/repositories/DrizzleCollectionRepository.ts` 288 288 + 289 289 + **Lines 504-511:** DELETE all existing cards: 290 290 + 291 291 + ```typescript 292 292 + // Delete existing collaborators and card links 293 293 + await tx 294 294 + .delete(collectionCollaborators) 295 295 + .where(eq(collectionCollaborators.collectionId, collectionData.id)); 296 296 + 297 297 + await tx 298 298 + .delete(collectionCards) 299 299 + .where(eq(collectionCards.collectionId, collectionData.id)); 300 300 + ``` 301 301 + 302 302 + **Lines 514-529:** INSERT only what's in memory: 303 303 + 304 304 + ```typescript 305 305 + // Insert new collaborators 306 306 + if (collaborators.length > 0) { 307 307 + await tx.insert(collectionCollaborators).values(collaborators); 308 308 + } 309 309 + 310 310 + // Insert new card links with mapped published record IDs 311 311 + if (cardLinks.length > 0) { 312 312 + const cardLinksWithMappedRecords = cardLinks.map((link) => ({ 313 313 + ...link, 314 314 + publishedRecordId: link.publishedRecordId 315 315 + ? linkPublishedRecordMap.get(link.publishedRecordId) || 316 316 + link.publishedRecordId 317 317 + : undefined, 318 318 + })); 319 319 + 320 320 + await tx.insert(collectionCards).values(cardLinksWithMappedRecords); 321 321 + } 322 322 + ``` 323 323 + 324 324 + --- 325 325 + 326 326 + ## Race Condition Timeline 327 327 + 328 328 + Let's trace what happens when 3 concurrent transactions try to add different cards to Collection X: 329 329 + 330 330 + ### Initial State 331 331 + 332 332 + - Collection X exists in database 333 333 + - Collection X has 0 cards 334 334 + - 3 collection items need to be added: Card-A, Card-B, Card-C 335 335 + 336 336 + ### Time T0: All Transactions Start 337 337 + 338 338 + **Transaction 1:** Add Card-A to Collection X 339 339 + **Transaction 2:** Add Card-B to Collection X 340 340 + **Transaction 3:** Add Card-C to Collection X 341 341 + 342 342 + All start concurrently via `Promise.allSettled()`. 343 343 + 344 344 + ### Time T1: READ Phase 345 345 + 346 346 + **Transaction 1:** 347 347 + 348 348 + ```typescript 349 349 + // CardCollectionService.ts line 54-58 350 350 + const collectionResult = await this.collectionRepository.findById(collectionId); 351 351 + // Returns: Collection X with cardLinks = [] 352 352 + ``` 353 353 + 354 354 + **Transaction 2:** 355 355 + 356 356 + ```typescript 357 357 + // CardCollectionService.ts line 54-58 358 358 + const collectionResult = await this.collectionRepository.findById(collectionId); 359 359 + // Returns: Collection X with cardLinks = [] 360 360 + ``` 361 361 + 362 362 + **Transaction 3:** 363 363 + 364 364 + ```typescript 365 365 + // CardCollectionService.ts line 54-58 366 366 + const collectionResult = await this.collectionRepository.findById(collectionId); 367 367 + // Returns: Collection X with cardLinks = [] 368 368 + ``` 369 369 + 370 370 + **Result:** All 3 transactions see the same initial state (0 cards). 371 371 + 372 372 + ### Time T2: MODIFY Phase (In-Memory) 373 373 + 374 374 + **Transaction 1:** 375 375 + 376 376 + ```typescript 377 377 + // CardCollectionService.ts line 70-82 378 378 + collection.addCard(cardA.id, ...); 379 379 + // In-memory state: collection.cardLinks = [Card-A] 380 380 + ``` 381 381 + 382 382 + **Transaction 2:** 383 383 + 384 384 + ```typescript 385 385 + // CardCollectionService.ts line 70-82 386 386 + collection.addCard(cardB.id, ...); 387 387 + // In-memory state: collection.cardLinks = [Card-B] 388 388 + ``` 389 389 + 390 390 + **Transaction 3:** 391 391 + 392 392 + ```typescript 393 393 + // CardCollectionService.ts line 70-82 394 394 + collection.addCard(cardC.id, ...); 395 395 + // In-memory state: collection.cardLinks = [Card-C] 396 396 + ``` 397 397 + 398 398 + **Result:** Each transaction has modified its own in-memory copy independently. 399 399 + 400 400 + ### Time T3: WRITE Phase (Database Commits) 401 401 + 402 402 + **Transaction 1 Commits First:** 403 403 + 404 404 + ```typescript 405 405 + // DrizzleCollectionRepository.ts lines 504-529 406 406 + await tx.delete(collectionCards).where(eq(collectionCards.collectionId, 'X')); 407 407 + // Deletes: 0 rows 408 408 + await tx.insert(collectionCards).values([Card - A]); 409 409 + // Inserts: Card-A 410 410 + ``` 411 411 + 412 412 + **Database State:** Collection X has 1 card: [Card-A] 413 413 + 414 414 + **Transaction 2 Commits Second:** 415 415 + 416 416 + ```typescript 417 417 + // DrizzleCollectionRepository.ts lines 504-529 418 418 + await tx.delete(collectionCards).where(eq(collectionCards.collectionId, 'X')); 419 419 + // Deletes: 1 row (Card-A is deleted!) ⚠️ 420 420 + await tx.insert(collectionCards).values([Card - B]); 421 421 + // Inserts: Card-B 422 422 + ``` 423 423 + 424 424 + **Database State:** Collection X has 1 card: [Card-B] (Card-A was lost!) 425 425 + 426 426 + **Transaction 3 Commits Last:** 427 427 + 428 428 + ```typescript 429 429 + // DrizzleCollectionRepository.ts lines 504-529 430 430 + await tx.delete(collectionCards).where(eq(collectionCards.collectionId, 'X')); 431 431 + // Deletes: 1 row (Card-B is deleted!) ⚠️ 432 432 + await tx.insert(collectionCards).values([Card - C]); 433 433 + // Inserts: Card-C 434 434 + ``` 435 435 + 436 436 + **Database State:** Collection X has 1 card: [Card-C] (Card-A and Card-B were lost!) 437 437 + 438 438 + ### Final Result 439 439 + 440 440 + - **Expected:** Collection X has 3 cards: [Card-A, Card-B, Card-C] 441 441 + - **Actual:** Collection X has 1 card: [Card-C] 442 442 + - **Lost:** Card-A and Card-B were successfully added in memory but deleted by subsequent transactions 443 443 + 444 444 + ### Why This Happens 445 445 + 446 446 + The DELETE-all-then-INSERT-all pattern means: 447 447 + 448 448 + 1. Each transaction only knows about cards it added in memory 449 449 + 2. Each transaction deletes ALL cards currently in the database 450 450 + 3. Each transaction inserts only the cards from its in-memory state 451 451 + 4. Later commits overwrite earlier commits completely 452 452 + 453 453 + --- 454 454 + 455 455 + ## Missing Safeguards 456 456 + 457 457 + ### 1. No Database-Level Unique Constraint 458 458 + 459 459 + **File:** `drizzle/0000_shocking_fenris.sql` 460 460 + 461 461 + The migration file does NOT include a `UNIQUE(collection_id, card_id)` constraint on the `collection_cards` table, even though the test schema suggests one was intended. 462 462 + 463 463 + **File:** `src/modules/cards/tests/test-utils/createTestSchema.ts` (line 49) 464 464 + 465 465 + ```typescript 466 466 + sql`CREATE TABLE IF NOT EXISTS collection_cards ( 467 467 + id UUID PRIMARY KEY DEFAULT uuid_generate_v4(), 468 468 + collection_id UUID NOT NULL REFERENCES collections(id) ON DELETE CASCADE, 469 469 + card_id UUID NOT NULL REFERENCES cards(id) ON DELETE CASCADE, 470 470 + added_by TEXT NOT NULL, 471 471 + added_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(), 472 472 + via_card_id UUID REFERENCES cards(id), 473 473 + published_record_id UUID REFERENCES published_records(id), 474 474 + UNIQUE(collection_id, card_id) 475 475 + )`, 476 476 + ``` 477 477 + 478 478 + **Issue:** The production schema doesn't enforce uniqueness, allowing duplicates if the code doesn't prevent them. 479 479 + 480 480 + ### 2. No Row-Level Locking 481 481 + 482 482 + There's no `SELECT FOR UPDATE` in the repository's `findById` method to prevent concurrent reads of the same collection. 483 483 + 484 484 + **File:** `src/modules/cards/infrastructure/repositories/DrizzleCollectionRepository.ts` 485 485 + 486 486 + **Line 55-94:** Standard SELECT without locking: 487 487 + 488 488 + ```typescript 489 489 + async findById( 490 490 + collectionId: CollectionId, 491 491 + ): Promise<Result<Collection | null, Error>> { 492 492 + try { 493 493 + const collectionResult = await this.db 494 494 + .select({ 495 495 + id: collections.id, 496 496 + authorId: collections.authorId, 497 497 + // ... more fields 498 498 + }) 499 499 + .from(collections) 500 500 + .where(eq(collections.id, collectionId.getStringValue())) 501 501 + .limit(1); 502 502 + 503 503 + // No .for('update') clause! 504 504 + // ... 505 505 + } 506 506 + } 507 507 + ``` 508 508 + 509 509 + ### 3. No Optimistic Concurrency Control 510 510 + 511 511 + The `collections` table has no version column to detect concurrent modifications: 512 512 + 513 513 + **File:** `src/modules/cards/infrastructure/repositories/schema/collection.sql.ts` 514 514 + 515 515 + ```typescript 516 516 + export const collections = pgTable('collections', { 517 517 + id: uuid('id').defaultRandom().primaryKey(), 518 518 + authorId: text('author_id').notNull(), 519 519 + name: text('name').notNull(), 520 520 + description: text('description'), 521 521 + accessType: text('access_type').notNull(), 522 522 + cardCount: integer('card_count').notNull().default(0), 523 523 + createdAt: timestamp('created_at', { withTimezone: true }) 524 524 + .notNull() 525 525 + .defaultNow(), 526 526 + updatedAt: timestamp('updated_at', { withTimezone: true }) 527 527 + .notNull() 528 528 + .defaultNow(), 529 529 + publishedRecordId: uuid('published_record_id').references( 530 530 + () => publishedRecords.id, 531 531 + ), 532 532 + // No version column! 533 533 + }); 534 534 + ``` 535 535 + 536 536 + ### 4. No Idempotent Inserts 537 537 + 538 538 + The code doesn't use `INSERT ... ON CONFLICT` to make inserts idempotent and safe for concurrent execution. 539 539 + 540 540 + --- 541 541 + 542 542 + ## Solution Options 543 543 + 544 544 + ### Option 1: INSERT with ON CONFLICT (Recommended for Concurrency) 545 545 + 546 546 + **Pros:** 547 547 + 548 548 + - ✅ Allows true concurrent operations without blocking 549 549 + - ✅ Database-enforced correctness via unique constraint 550 550 + - ✅ Naturally idempotent - safe to retry 551 551 + - ✅ Best performance for high-concurrency scenarios 552 552 + 553 553 + **Cons:** 554 554 + 555 555 + - ⚠️ Requires schema migration (add unique constraint) 556 556 + - ⚠️ More complex repository logic to track additions vs. removals 557 557 + 558 558 + **Implementation:** 559 559 + 560 560 + 1. **Add unique constraint to schema:** 561 561 + 562 562 + **File:** `src/modules/cards/infrastructure/repositories/schema/collection.sql.ts` 563 563 + 564 564 + ```typescript 565 565 + export const collectionCards = pgTable( 566 566 + 'collection_cards', 567 567 + { 568 568 + // ... existing columns ... 569 569 + }, 570 570 + (table) => ({ 571 571 + uniqueCollectionCard: unique().on(table.collectionId, table.cardId), 572 572 + }), 573 573 + ); 574 574 + ``` 575 575 + 576 576 + 2. **Generate migration:** `npm run db:generate` 577 577 + 578 578 + 3. **Change repository save method:** 579 579 + 580 580 + **File:** `src/modules/cards/infrastructure/repositories/DrizzleCollectionRepository.ts` 581 581 + 582 582 + ```typescript 583 583 + // Instead of DELETE + INSERT, use INSERT ON CONFLICT 584 584 + if (cardLinks.length > 0) { 585 585 + await tx 586 586 + .insert(collectionCards) 587 587 + .values(cardLinksWithMappedRecords) 588 588 + .onConflictDoUpdate({ 589 589 + target: [collectionCards.collectionId, collectionCards.cardId], 590 590 + set: { 591 591 + publishedRecordId: sql`EXCLUDED.published_record_id`, 592 592 + addedAt: sql`EXCLUDED.added_at`, 593 593 + addedBy: sql`EXCLUDED.added_by`, 594 594 + viaCardId: sql`EXCLUDED.via_card_id`, 595 595 + }, 596 596 + }); 597 597 + } 598 598 + ``` 599 599 + 600 600 + 4. **Track and delete removed cards separately** (if needed for card removal functionality) 601 601 + 602 602 + --- 603 603 + 604 604 + ### Option 2: SELECT FOR UPDATE Locking (Simplest but Serializes) 605 605 + 606 606 + **Pros:** 607 607 + 608 608 + - ✅ Simple to implement 609 609 + - ✅ No schema changes needed 610 610 + - ✅ Guarantees serial access to collections 611 611 + 612 612 + **Cons:** 613 613 + 614 614 + - ⚠️ Serializes all updates to the same collection (slower) 615 615 + - ⚠️ Potential for deadlocks if not careful 616 616 + - ⚠️ Reduced concurrency 617 617 + 618 618 + **Implementation:** 619 619 + 620 620 + **File:** `src/modules/cards/infrastructure/repositories/DrizzleCollectionRepository.ts` 621 621 + 622 622 + In `findById` method, add row-level locking: 623 623 + 624 624 + ```typescript 625 625 + async findById( 626 626 + collectionId: CollectionId, 627 627 + ): Promise<Result<Collection | null, Error>> { 628 628 + try { 629 629 + const collectionResult = await this.db 630 630 + .select({...}) 631 631 + .from(collections) 632 632 + .where(eq(collections.id, collectionId.getStringValue())) 633 633 + .for('update') // ← Add this line 634 634 + .limit(1); 635 635 + // ... 636 636 + } 637 637 + } 638 638 + ``` 639 639 + 640 640 + This forces concurrent transactions to wait in line, preventing the race condition but reducing throughput. 641 641 + 642 642 + --- 643 643 + 644 644 + ### Option 3: Application-Level Grouping (No DB Changes) 645 645 + 646 646 + **Pros:** 647 647 + 648 648 + - ✅ No database schema changes 649 649 + - ✅ No repository changes 650 650 + 651 651 + **Cons:** 652 652 + 653 653 + - ⚠️ Requires changes in sync use case 654 654 + - ⚠️ More complex application logic 655 655 + - ⚠️ Doesn't prevent races from other code paths (e.g., firehose) 656 656 + 657 657 + **Implementation:** 658 658 + 659 659 + **File:** `src/modules/sync/application/useCases/SyncAccountDataUseCase.ts` 660 660 + 661 661 + Group collection items by collection before processing: 662 662 + 663 663 + ```typescript 664 664 + private async syncCollection(...): Promise<number> { 665 665 + // ... fetch records ... 666 666 + 667 667 + // Group items by collection 668 668 + const itemsByCollection = new Map<string, any[]>(); 669 669 + for (const record of records) { 670 670 + const collectionId = extractCollectionId(record); // Helper function 671 671 + if (!itemsByCollection.has(collectionId)) { 672 672 + itemsByCollection.set(collectionId, []); 673 673 + } 674 674 + itemsByCollection.get(collectionId)!.push(record); 675 675 + } 676 676 + 677 677 + // Process each collection's items sequentially 678 678 + for (const [collectionId, items] of itemsByCollection) { 679 679 + for (const item of items) { 680 680 + await this.processRecord(item, useCase, curatorId, collectionType); 681 681 + } 682 682 + } 683 683 + } 684 684 + ``` 685 685 + 686 686 + --- 687 687 + 688 688 + ### Option 4: Optimistic Concurrency Control (Version Column) 689 689 + 690 690 + **Pros:** 691 691 + 692 692 + - ✅ Detects conflicts explicitly 693 693 + - ✅ No blocking/locking 694 694 + 695 695 + **Cons:** 696 696 + 697 697 + - ⚠️ Requires retry logic and error handling 698 698 + - ⚠️ Schema changes needed (version column) 699 699 + - ⚠️ More complex to implement correctly 700 700 + 701 701 + **Implementation:** Add a `version` column to collections table and check it during updates. 702 702 + 703 703 + --- 704 704 + 705 705 + ## Recommended Approach 706 706 + 707 707 + **Use Option 1: INSERT with ON CONFLICT** 708 708 + 709 709 + This is the most robust solution for a concurrent, distributed system: 710 710 + 711 711 + 1. **Add UNIQUE constraint** to `collection_cards` table 712 712 + 2. **Change save method** to use `INSERT ... ON CONFLICT` 713 713 + 3. **Maintain concurrent processing** in sync use case 714 714 + 4. **Ensure idempotency** across all code paths 715 715 + 716 716 + **Why this is best:** 717 717 + 718 718 + - Handles concurrent operations from both sync AND firehose 719 719 + - Database enforces correctness (no silent data loss) 720 720 + - Scales well with high concurrency 721 721 + - Aligns with the domain model (Collection already prevents duplicate cards) 722 722 + 723 723 + --- 724 724 + 725 725 + ## Additional Notes 726 726 + 727 727 + ### Why Tests Didn't Catch This 728 728 + 729 729 + **File:** `src/modules/cards/tests/test-utils/createTestSchema.ts` (line 49) 730 730 + 731 731 + The test schema DOES include `UNIQUE(collection_id, card_id)`, which would have helped catch this issue: 732 732 + 733 733 + ```typescript 734 734 + UNIQUE(collection_id, card_id); 735 735 + ``` 736 736 + 737 737 + However, the production migration doesn't have this constraint, creating a gap between test and production schemas. 738 738 + 739 739 + ### Schema Drift 740 740 + 741 741 + There's a discrepancy between: 742 742 + 743 743 + - **Production schema:** `drizzle/0000_shocking_fenris.sql` (no unique constraint) 744 744 + - **Test schema:** `createTestSchema.ts` (has unique constraint) 745 745 + 746 746 + This drift should be resolved to ensure test environments match production. 747 747 + 748 748 + --- 749 749 + 750 750 + ## Conclusion 751 751 + 752 752 + The root cause is a **lost update race condition** caused by: 753 753 + 754 754 + 1. Concurrent processing of collection items 755 755 + 2. Read-Modify-Write pattern without concurrency control 756 756 + 3. DELETE-all-then-INSERT-all repository save pattern 757 757 + 758 758 + The recommended fix is to use **INSERT with ON CONFLICT**, which provides: 759 759 + 760 760 + - Database-level concurrency safety 761 761 + - Idempotent operations 762 762 + - Maximum throughput for concurrent operations 763 763 + 764 764 + Implementation requires: 765 765 + 766 766 + - Schema migration to add unique constraint 767 767 + - Repository method refactoring to use ON CONFLICT 768 768 + - Alignment of test and production schemas

+32 -7

src/modules/sync/application/useCases/SyncAccountDataUseCase.ts

View file Reviewed

··· 180 180 console.log(`[SYNC] Processed ${collectionsProcessed} collections`); 181 181 182 182 // 3. Sync collection items (links cards to collections) 183 183 + // NOTE: Process sequentially to prevent race conditions when multiple items 184 184 + // are added to the same collection concurrently 183 185 console.log(`[SYNC] Syncing collection items for ${curatorId}...`); 184 186 const itemsProcessed = await this.syncCollection( 185 187 curatorId, 186 188 ATPROTO_NSID.MARGIN.COLLECTION_ITEM, 187 189 this.processMarginCollectionItemUseCase, 190 190 + true, // sequential = true 188 191 ); 189 192 totalProcessed += itemsProcessed; 190 193 console.log(`[SYNC] Processed ${itemsProcessed} collection items`); ··· 194 197 195 198 /** 196 199 * Sync a specific collection type with batched concurrent processing 200 200 + * @param sequential - If true, process records sequentially instead of concurrently (prevents race conditions) 197 201 */ 198 202 private async syncCollection( 199 203 curatorId: string, ··· 202 206 | ProcessMarginBookmarkFirehoseEventUseCase 203 207 | ProcessMarginCollectionFirehoseEventUseCase 204 208 | ProcessMarginCollectionItemFirehoseEventUseCase, 209 209 + sequential: boolean = false, 205 210 ): Promise<number> { 206 211 let totalProcessed = 0; 207 212 let pageCount = 0; ··· 234 239 const batchNumber = Math.floor(i / CONCURRENT_BATCH_SIZE) + 1; 235 240 236 241 console.log( 237 237 - `[SYNC] Processing batch ${batchNumber} (${batch.length} records) from page ${pageCount}...`, 242 242 + `[SYNC] Processing batch ${batchNumber} (${batch.length} records) from page ${pageCount}${sequential ? ' [SEQUENTIAL]' : ''}...`, 238 243 ); 239 244 240 240 - // Process all records in this batch concurrently 241 241 - const results = await Promise.allSettled( 242 242 - batch.map((record) => 243 243 - this.processRecord(record, useCase, curatorId, collectionType), 244 244 - ), 245 245 - ); 245 245 + let results: PromiseSettledResult<boolean>[]; 246 246 + 247 247 + if (sequential) { 248 248 + // Process records one at a time to prevent race conditions 249 249 + results = []; 250 250 + for (const record of batch) { 251 251 + try { 252 252 + const result = await this.processRecord( 253 253 + record, 254 254 + useCase, 255 255 + curatorId, 256 256 + collectionType, 257 257 + ); 258 258 + results.push({ status: 'fulfilled', value: result }); 259 259 + } catch (error) { 260 260 + results.push({ status: 'rejected', reason: error }); 261 261 + } 262 262 + } 263 263 + } else { 264 264 + // Process all records in this batch concurrently 265 265 + results = await Promise.allSettled( 266 266 + batch.map((record) => 267 267 + this.processRecord(record, useCase, curatorId, collectionType), 268 268 + ), 269 269 + ); 270 270 + } 246 271 247 272 // Count successful processes 248 273 const successCount = results.filter(