Skip to content

chore: limit staging area processing by 10 tries #278

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rodrigok merged 1 commit into from
Oct 12, 2025
Merged

chore: limit staging area processing by 10 tries #278

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions packages/core/src/models/event.model.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ export interface EventStore<E = Pdu> extends PersistentEventBase<E> {

export interface EventStagingStore extends PersistentEventBase {
roomId: string;
got: number;
from: 'join' | 'transaction';
}

Expand Down
21 changes: 18 additions & 3 deletions packages/federation-sdk/src/repositories/event-staging.repository.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,12 @@ export class EventStagingRepository {
@inject('EventStagingCollection')
private readonly collection: Collection<EventStagingStore>,
) {
this.collection.createIndex({ roomId: 1, createdAt: 1 });
this.collection.createIndex({
roomId: 1,
got: 1,
'event.depth': 1,
createdAt: 1,
});
}

async create(
Expand All @@ -30,6 +35,7 @@ export class EventStagingRepository {
$setOnInsert: {
roomId: event.room_id,
createdAt: new Date(),
got: 0,
},
$set: {
event,
Expand All @@ -43,9 +49,18 @@ export class EventStagingRepository {
}

getLeastDepthEventForRoom(roomId: string): Promise<EventStagingStore | null> {
return this.collection.findOne(
return this.collection.findOneAndUpdate(
{ roomId },
{ sort: { 'event.depth': 1, createdAt: 1 } },
{
$inc: {
got: 1,
},
},
{
sort: { got: 1, 'event.depth': 1, createdAt: 1 },
upsert: false,
returnDocument: 'before',
},
);
}

Expand Down
2 changes: 1 addition & 1 deletion packages/federation-sdk/src/server-discovery/discovery.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ const SERVER_DISCOVERY_CACHE_MAX_AGE =
}

throw new Error('Invalid SERVER_DISCOVERY_CACHE_MAX_AGE value');
})(process.env.SERVER_DISCOVERY_CACHE_MAX_AGE) ?? 3_600_000; // default to 1 hour
})(process.env.SERVER_DISCOVERY_CACHE_MAX_AGE) ?? 300_000; // default to 5 minutes

// should only be needed if input is from a dns server
function fix6(addr: string): `[${string}]` {
Expand Down
68 changes: 39 additions & 29 deletions packages/federation-sdk/src/services/staging-area.service.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,18 @@ import { FederationService } from './federation.service';
import { MissingEventService } from './missing-event.service';
import { PartialStateResolutionError, StateService } from './state.service';

const MAX_EVENT_RETRY =
((maxRetry?: string) => {
if (!maxRetry) return;

const n = Number.parseInt(maxRetry, 10);
if (!Number.isNaN(n) && n >= 0) {
return n;
}

throw new Error('Invalid MAX_EVENT_RETRY value');
})(process.env.MAX_EVENT_RETRY) ?? 10;

class MissingAuthorizationEventsError extends Error {
constructor(message: string) {
super(message);
Expand Down Expand Up @@ -62,16 +74,6 @@ export class StagingAreaService {
}

async processEventForRoom(roomId: string) {
let event = await this.eventService.getLeastDepthEventForRoom(roomId);
if (!event) {
this.logger.debug({ msg: 'No staged event found for room', roomId });
await this.lockRepository.releaseLock(
roomId,
this.configService.instanceId,
);
return;
}

const roomIdToRoomVersion = new Map<string, RoomVersion>();
const getRoomVersion = async (roomId: string) => {
if (roomIdToRoomVersion.has(roomId)) {
Expand All @@ -88,9 +90,32 @@ export class StagingAreaService {
return PersistentEventFactory.createFromRawEvent(pdu, version);
};

while (event) {
let event: EventStagingStore | null = null;

do {
event = await this.eventService.getLeastDepthEventForRoom(roomId);
if (!event) {
this.logger.debug({ msg: 'No staged event found for room', roomId });
break;
}

if (event.got > MAX_EVENT_RETRY) {
this.logger.warn(
`Event ${event._id} has been tried ${MAX_EVENT_RETRY} times, removing from staging area`,
);
await this.eventService.markEventAsUnstaged(event);
continue;
}
Comment on lines +102 to +108

@coderabbitai coderabbitai Bot Oct 12, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Potential off-by-one error in retry limit.

With got starting at 0 and findOneAndUpdate returning the document before the increment (due to returnDocument: 'before'), the current comparison event.got > MAX_EVENT_RETRY allows 11 processing attempts (when got is 0 through 10) instead of 10.

If the intent is to limit processing to exactly MAX_EVENT_RETRY attempts, change the comparison to >=:

-			if (event.got > MAX_EVENT_RETRY) {
+			if (event.got >= MAX_EVENT_RETRY) {

Additionally, the log message hardcodes "10" and should use the variable:

 				this.logger.warn(
-					`Event ${event._id} has been tried more than 10 times, removing from staging area`,
+					`Event ${event._id} has been tried ${MAX_EVENT_RETRY} times, removing from staging area`,
 				);
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
if (event.got > MAX_EVENT_RETRY) {
this.logger.warn(
`Event ${event._id} has been tried more than 10 times, removing from staging area`,
);
await this.eventService.markEventAsUnstaged(event);
continue;
}
if (event.got >= MAX_EVENT_RETRY) {
this.logger.warn(
`Event ${event._id} has been tried ${MAX_EVENT_RETRY} times, removing from staging area`,
);
await this.eventService.markEventAsUnstaged(event);
continue;
}


this.logger.info({ msg: 'Processing event', eventId: event._id });

// if we got an event, we need to update the lock's timestamp to avoid it being timed out
// and acquired by another instance while we're processing a batch of events for this room
await this.lockRepository.updateLockTimestamp(
roomId,
this.configService.instanceId,
);

try {
const addedMissing = await this.processDependencyStage(event);
if (addedMissing) {
Expand Down Expand Up @@ -123,33 +148,18 @@ export class StagingAreaService {
});
} else if (err instanceof MissingEventsError) {
this.logger.info({
msg: 'Added missing events, postponing current event processing',
msg: 'Added missing events, postponing event processing',
eventId: event._id,
});
} else {
this.logger.error({
msg: 'Error processing event',
msg: 'Error processing event, postponing event processing',
event,
err,
});

await this.eventService.markEventAsUnstaged(event);
}
}

// TODO: what should we do to avoid infinite loops in case the next event is always the same event

event = await this.eventService.getLeastDepthEventForRoom(roomId);

// if we got an event, we need to update the lock's timestamp to avoid it being timed out
// and acquired by another instance while we're processing a batch of events for this room
if (event) {
await this.lockRepository.updateLockTimestamp(
roomId,
this.configService.instanceId,
);
}
}
} while (event);

// release the lock after processing
await this.lockRepository.releaseLock(
Expand Down