nandi.uk
This repository has no description
formatting
author
Wesley Finck
date
(Oct 23, 2025, 12:49 PM -0700)
commit
9738a59d
9738a59d104c1dfd0adc11299f5a0ea566281360
parent
62ef839d
62ef839dc84ea5271d34e28f0bcc87dc952bfb5b
+40
-20
+31
-12
src/modules/cards/tests/integration/BullMQEventSystem.integration.test.ts
+31
-12
src/modules/cards/tests/integration/BullMQEventSystem.integration.test.ts
···
350
350
// Create events for same card/user (will compete for same lock)
351
351
const cardId = CardId.createFromString('concurrent-test-card').unwrap();
352
352
const curatorId = CuratorId.create('did:plc:concurrentuser').unwrap();
353
-
const collectionId1 = CollectionId.createFromString('collection-1').unwrap();
354
-
const collectionId2 = CollectionId.createFromString('collection-2').unwrap();
353
+
const collectionId1 =
354
+
CollectionId.createFromString('collection-1').unwrap();
355
+
const collectionId2 =
356
+
CollectionId.createFromString('collection-2').unwrap();
355
357
356
358
const events = [
357
359
CardAddedToLibraryEvent.create(cardId, curatorId).unwrap(),
358
-
CardAddedToCollectionEvent.create(cardId, collectionId1, curatorId).unwrap(),
359
-
CardAddedToCollectionEvent.create(cardId, collectionId2, curatorId).unwrap(),
360
+
CardAddedToCollectionEvent.create(
361
+
cardId,
362
+
collectionId1,
363
+
curatorId,
364
+
).unwrap(),
365
+
CardAddedToCollectionEvent.create(
366
+
cardId,
367
+
collectionId2,
368
+
curatorId,
369
+
).unwrap(),
360
370
];
361
371
362
372
// Act - Process all events concurrently (not sequentially)
363
373
const results = await Promise.all([
364
-
saga1.handleCardEvent(events[0]),
365
-
saga2.handleCardEvent(events[1]),
366
-
saga3.handleCardEvent(events[2]),
374
+
saga1.handleCardEvent(events[0]!),
375
+
saga2.handleCardEvent(events[1]!),
376
+
saga3.handleCardEvent(events[2]!),
367
377
]);
368
378
369
379
// Assert - All should succeed (no events dropped due to lock contention)
···
397
407
// Create 10 collection events that will all compete for the same lock
398
408
const events = Array.from({ length: 10 }, (_, i) => {
399
409
const saga = new CardCollectionSaga(mockUseCase, stateStore);
400
-
const collectionId = CollectionId.createFromString(`collection-${i}`).unwrap();
401
-
const event = CardAddedToCollectionEvent.create(cardId, collectionId, curatorId).unwrap();
410
+
const collectionId = CollectionId.createFromString(
411
+
`collection-${i}`,
412
+
).unwrap();
413
+
const event = CardAddedToCollectionEvent.create(
414
+
cardId,
415
+
collectionId,
416
+
curatorId,
417
+
).unwrap();
402
418
return { saga, event };
403
419
});
404
420
405
421
// Add one library event
406
422
const librarySaga = new CardCollectionSaga(mockUseCase, stateStore);
407
-
const libraryEvent = CardAddedToLibraryEvent.create(cardId, curatorId).unwrap();
423
+
const libraryEvent = CardAddedToLibraryEvent.create(
424
+
cardId,
425
+
curatorId,
426
+
).unwrap();
408
427
409
428
// Act - Process all events concurrently
410
429
const allPromises = [
···
426
445
expect(call.cardId).toBe(cardId.getStringValue());
427
446
expect(call.actorId).toBe(curatorId.value);
428
447
expect(call.collectionIds).toHaveLength(10);
429
-
448
+
430
449
// Verify all collection IDs are present
431
450
for (let i = 0; i < 10; i++) {
432
451
expect(call.collectionIds).toContain(`collection-${i}`);
···
492
511
493
512
// Act - Start first saga (will acquire lock)
494
513
const promise1 = lockHoldingSaga.handleCardEvent(event1);
495
-
514
+
496
515
// Immediately start second saga (will need to retry)
497
516
const promise2 = retryingSaga.handleCardEvent(event2);
498
517
+4
-5
src/modules/feeds/application/sagas/CardCollectionSaga.ts
+4
-5
src/modules/feeds/application/sagas/CardCollectionSaga.ts
···
29
29
async handleCardEvent(
30
30
event: CardAddedToLibraryEvent | CardAddedToCollectionEvent,
31
31
): Promise<Result<void>> {
32
-
console.log('Handling card event:', event);
33
32
const aggregationKey = this.createKey(event);
34
33
35
34
// Retry lock acquisition with longer delays and more attempts for high concurrency
···
45
44
const existing = await this.getPendingActivity(aggregationKey);
46
45
47
46
if (existing && this.isWithinWindow(existing)) {
48
-
console.log(`Merging event into existing activity for ${aggregationKey}`);
49
47
this.mergeActivity(existing, event);
50
48
await this.setPendingActivity(aggregationKey, existing);
51
49
} else {
52
-
console.log(`Creating new pending activity for ${aggregationKey}`);
53
50
const newActivity = this.createNewPendingActivity(event);
54
51
await this.setPendingActivity(aggregationKey, newActivity);
55
52
await this.scheduleFlush(aggregationKey);
···
67
64
const exponentialDelay = baseDelay * Math.pow(1.5, attempt);
68
65
const jitter = Math.random() * 50; // Add randomness to prevent thundering herd
69
66
const delay = Math.min(exponentialDelay + jitter, maxDelay);
70
-
71
-
console.log(`Lock acquisition failed for ${aggregationKey}, retrying in ${Math.round(delay)}ms (attempt ${attempt + 1}/${maxRetries})`);
67
+
68
+
console.log(
69
+
`Lock acquisition failed for ${aggregationKey}, retrying in ${Math.round(delay)}ms (attempt ${attempt + 1}/${maxRetries})`,
70
+
);
72
71
await new Promise((resolve) => setTimeout(resolve, delay));
73
72
}
74
73
}
+5
-3
src/workers/feed-worker.ts
+5
-3
src/workers/feed-worker.ts
···
3
3
4
4
async function main() {
5
5
const useInMemoryEvents = process.env.USE_IN_MEMORY_EVENTS === 'true';
6
-
6
+
7
7
if (useInMemoryEvents) {
8
-
console.log('Skipping feed worker startup - using in-memory events (handled by main process)');
8
+
console.log(
9
+
'Skipping feed worker startup - using in-memory events (handled by main process)',
10
+
);
9
11
return;
10
12
}
11
13
12
14
console.log('Starting dedicated feed worker process...');
13
15
const configService = new EnvironmentConfigService();
14
16
const feedWorkerProcess = new FeedWorkerProcess(configService);
15
-
17
+
16
18
await feedWorkerProcess.start();
17
19
}
18
20