feat: partial states (basic) #267
Conversation
WalkthroughAdds a per-event Changes
Sequence Diagram(s)sequenceDiagram
autonumber
participant Client as Incoming
participant RS as RoomService
participant SS as StateService
participant EF as EventFetcherService
participant ER as EventRepository
participant SGR as StateGraphRepo
rect rgba(230,245,255,0.6)
note over Client,RS: Join / incoming PDU with partial-state-aware resolution
Client->>RS: incoming join PDU
RS->>SS: handlePdu(pdu)
SS->>SS: isRoomStatePartial(roomId)?
alt state is partial
SS->>EF: fetch missing events (walk branches)
EF-->>SS: missing PDUs
SS->>ER: insertOrUpdateEventWithStateId(..., partial=true)
SS->>SGR: createDelta(..., partial=true)
SS-->>RS: throw/postpone (PartialStateResolutionError)
else state not partial
SS->>ER: insertOrUpdateEventWithStateId(..., partial=false)
SS->>SGR: createDelta(..., partial=false)
SS-->>RS: success / persisted
end
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Suggested reviewers
Poem
Pre-merge checks and finishing touches✅ Passed checks (3 passed)
✨ Finishing touches
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
debdutdeb
force-pushed
the
feat/partial-states
branch
from
f984b0e to
c72665b
Compare
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #267 +/- ##
==========================================
- Coverage 62.34% 60.15% -2.19%
==========================================
Files 67 67
Lines 6393 6644 +251
==========================================
+ Hits 3986 3997 +11
- Misses 2407 2647 +240 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
packages/federation-sdk/src/services/state.service.ts (1)
247-265: Preservepartialwhen rehydrating eventsWe persist the
partialflag onEventStore, but both_getStore().getEventsandbuildStateFromStateMapinstantiate wrappers without forwardingevent.partial. Any event read from Mongo is thus re-created withpartial === false, so callers lose the signal that a state node is incomplete (e.g.,getLatestRoomState,_getStoreconsumers). Pass the stored flag intoPersistentEventFactory.createFromRawEventin both places.- const e = PersistentEventFactory.createFromRawEvent( - event.event, - roomVersion, - ); + const e = PersistentEventFactory.createFromRawEvent( + event.event, + roomVersion, + event.partial ?? false, + );Do the same inside
buildStateFromStateMapwhen hydratingevent.Also applies to: 360-367
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
Disabled knowledge base sources:
- Jira integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (7)
packages/core/src/models/event.model.ts(1 hunks)packages/federation-sdk/src/repositories/event.repository.ts(5 hunks)packages/federation-sdk/src/repositories/state-graph.repository.ts(3 hunks)packages/federation-sdk/src/services/state.service.spec.ts(5 hunks)packages/federation-sdk/src/services/state.service.ts(8 hunks)packages/room/src/manager/event-wrapper.ts(3 hunks)packages/room/src/manager/factory.ts(2 hunks)
| afterEach(async () => { | ||
| await Promise.all([ | ||
| eventCollection.deleteMany(), | ||
| stateGraphCollection.deleteMany(), | ||
| ]); | ||
| }); |
There was a problem hiding this comment.
Fix Mongo cleanup filter
deleteMany requires an explicit filter; calling it without one throws (filter parameter must be an object). The cleanup hook will fail before every test run. Pass an empty filter to wipe the collections.
- await Promise.all([
- eventCollection.deleteMany(),
- stateGraphCollection.deleteMany(),
- ]);
+ await Promise.all([
+ eventCollection.deleteMany({}),
+ stateGraphCollection.deleteMany({}),
+ ]);📝 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.
| afterEach(async () => { | |
| await Promise.all([ | |
| eventCollection.deleteMany(), | |
| stateGraphCollection.deleteMany(), | |
| ]); | |
| }); | |
| afterEach(async () => { | |
| await Promise.all([ | |
| eventCollection.deleteMany({}), | |
| stateGraphCollection.deleteMany({}), | |
| ]); | |
| }); |
🤖 Prompt for AI Agents
In packages/federation-sdk/src/services/state.service.spec.ts around lines 143
to 148, the afterEach cleanup calls deleteMany() without a filter which throws;
change both calls to pass an explicit empty filter object (e.g., {}) so the
collections are wiped correctly; update the two deleteMany invocations to
deleteMany({}) to fix the failing cleanup hook.
ggazzo
force-pushed
the
feat/partial-states
branch
2 times, most recently
from
fd0a8cf to
3a310a3
Compare
ggazzo
force-pushed
the
feat/partial-states
branch
from
3a310a3 to
a7547ba
Compare
ggazzo
force-pushed
the
feat/partial-states
branch
from
a7547ba to
cd7ff05
Compare
There was a problem hiding this comment.
Actionable comments posted: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
packages/federation-sdk/src/services/invite.service.ts (1)
85-91: Avoid passing empty string to extractDomainFromId; add explicit guardextractDomainFromId('') throws. Guard stateKey first; then extract without nullish fallback.
- const invitedServer = extractDomainFromId(inviteEvent.stateKey ?? ''); - if (!invitedServer) { - throw new Error( - `invalid state_key ${inviteEvent.stateKey}, no server_name part`, - ); - } + if (!inviteEvent.stateKey) { + throw new Error('failed to invite: invite event missing state_key'); + } + const invitedServer = extractDomainFromId(inviteEvent.stateKey);
♻️ Duplicate comments (1)
packages/federation-sdk/src/services/state.service.spec.ts (1)
142-147: Fix MongoDB deleteMany filter (duplicate concern).The
deleteMany()calls require an explicit filter parameter. Calling them without arguments will throw a MongoDB error: "filter parameter must be an object."This is a duplicate of a previous review comment that was not addressed.
Apply this fix:
beforeEach(async () => { await Promise.all([ - eventCollection.deleteMany(), - stateGraphCollection.deleteMany(), + eventCollection.deleteMany({}), + stateGraphCollection.deleteMany({}), ]); });
🧹 Nitpick comments (7)
packages/federation-sdk/src/services/federation.service.ts (2)
260-269: Avoid re-signing per destinationsignEvent is independent of destination. Sign once before the loop to avoid repeated work.
- // TODO: signing should happen here over local persisting - // should be handled in transaction queue implementation - await this.stateService.signEvent(event); + // signing should happen once before sending to all serversAdd once before the for-of:
await this.stateService.signEvent(event); for (const server of servers) { // ... }
271-283: Reduce log verbosity; avoid logging full transaction payloadLogging full txn (including PDUs) can be large and contain sensitive content. Log IDs and counts instead.
- this.logger.info({ - transaction: txn, - msg: `Sending event ${event.eventId} to server: ${server}`, - }); + this.logger.info({ + msg: `Sending event ${event.eventId} to server: ${server}`, + pduCount: txn.pdus.length, + eduCount: txn.edus.length, + });packages/federation-sdk/src/services/invite.service.ts (1)
16-16: Remove unused importUnknownRoomError isn’t used here.
-import { StateService, UnknownRoomError } from './state.service'; +import { StateService } from './state.service';packages/federation-sdk/src/services/room.service.ts (4)
1044-1046: Add error messages for thrown ErrorsThrowing plain Error() hampers diagnosis; include context.
- throw new Error(); + throw new Error(`join: insufficient servers to fetch missing events for branch ${context.event.eventId} in ${roomId}`); @@ - throw new Error(); + throw new Error(`join: exhausted server list; still missing events ${missing.join(', ')} for branch ${context.event.eventId} in ${roomId}`);Also applies to: 1072-1074
899-937: Avoid shadowing variable “event” inside loopThe inner for (const event of partialEvents) shadows the outer “event” from send_join response. Rename for clarity and to prevent accidental misuse.
- for (const event of partialEvents) { - this.logger.info({ roomId, eventId: event.eventId }, 'walking branch'); + for (const partialEvent of partialEvents) { + this.logger.info({ roomId, eventId: partialEvent.eventId }, 'walking branch'); @@ - { event }, + { event: partialEvent }, @@ - eventId: event.eventId, - depth: event.depth, - previous: event.getPreviousEventIds(), + eventId: partialEvent.eventId, + depth: partialEvent.depth, + previous: partialEvent.getPreviousEventIds(), @@ - await stateService._resolveStateAtEvent(missingEvent); + await stateService._resolveStateAtEvent(missingEvent); @@ - await stateService._resolveStateAtEvent(event); + await stateService._resolveStateAtEvent(partialEvent);
975-979: Reduce logging of full event payloadLog identifiers and minimal metadata to avoid large/PII logs.
- logger.info({ - msg: 'Persisting join event', - eventId: joinEventFinal.eventId, - event: joinEventFinal.event, - }); + logger.info({ + msg: 'Persisting join event', + eventId: joinEventFinal.eventId, + roomId: joinEventFinal.roomId, + depth: joinEventFinal.depth, + });
843-851: Prefer defaultRoomVersion constantUse PersistentEventFactory.defaultRoomVersion instead of a hardcoded '10' to keep consistency with supported versions.
- const roomVersion = '10' as const; + const roomVersion = PersistentEventFactory.defaultRoomVersion;
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
Disabled knowledge base sources:
- Jira integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (14)
packages/core/src/models/event.model.ts(1 hunks)packages/federation-sdk/src/repositories/event.repository.ts(5 hunks)packages/federation-sdk/src/repositories/state-graph.repository.ts(3 hunks)packages/federation-sdk/src/services/event-fetcher.service.ts(1 hunks)packages/federation-sdk/src/services/federation.service.ts(2 hunks)packages/federation-sdk/src/services/invite.service.ts(6 hunks)packages/federation-sdk/src/services/profiles.service.ts(1 hunks)packages/federation-sdk/src/services/room.service.ts(5 hunks)packages/federation-sdk/src/services/send-join.service.ts(1 hunks)packages/federation-sdk/src/services/staging-area.service.ts(2 hunks)packages/federation-sdk/src/services/state.service.spec.ts(4 hunks)packages/federation-sdk/src/services/state.service.ts(13 hunks)packages/room/src/manager/event-wrapper.ts(3 hunks)packages/room/src/manager/factory.ts(2 hunks)
🚧 Files skipped from review as they are similar to previous changes (3)
- packages/room/src/manager/factory.ts
- packages/room/src/manager/event-wrapper.ts
- packages/federation-sdk/src/services/state.service.ts
🧰 Additional context used
🧬 Code graph analysis (7)
packages/federation-sdk/src/services/send-join.service.ts (1)
packages/room/src/manager/event-wrapper.ts (1)
event(129-138)
packages/federation-sdk/src/services/room.service.ts (4)
packages/federation-sdk/src/services/state.service.ts (2)
singleton(64-1175)UnknownRoomError(58-62)packages/federation-sdk/src/services/event-fetcher.service.ts (1)
singleton(15-143)packages/core/src/index.ts (2)
createLogger(79-79)logger(79-79)packages/room/src/manager/event-wrapper.ts (3)
extractDomainFromId(23-29)roomId(101-103)event(129-138)
packages/federation-sdk/src/services/staging-area.service.ts (1)
packages/federation-sdk/src/services/state.service.ts (1)
PartialStateResolutionError(51-56)
packages/federation-sdk/src/services/federation.service.ts (1)
packages/room/src/manager/event-wrapper.ts (2)
event(129-138)extractDomainFromId(23-29)
packages/federation-sdk/src/services/invite.service.ts (2)
packages/room/src/manager/event-wrapper.ts (2)
roomId(101-103)extractDomainFromId(23-29)packages/room/src/manager/factory.ts (1)
PersistentEventFactory(30-123)
packages/federation-sdk/src/services/state.service.spec.ts (4)
packages/federation-sdk/src/index.ts (4)
EventID(27-27)PersistentEventBase(25-25)EventStore(32-32)UserID(28-28)packages/room/src/types/_common.ts (2)
EventID(8-8)UserID(20-20)packages/room/src/state_resolution/definitions/definitions.ts (1)
EventStore(97-99)packages/room/src/manager/factory.ts (1)
PersistentEventFactory(30-123)
packages/federation-sdk/src/repositories/event.repository.ts (3)
packages/room/src/manager/event-wrapper.ts (1)
roomId(101-103)packages/federation-sdk/src/index.ts (1)
RoomID(29-29)packages/room/src/types/_common.ts (1)
RoomID(16-16)
🔇 Additional comments (17)
packages/federation-sdk/src/repositories/state-graph.repository.ts (2)
25-25: LGTM: Partial flag added to StateGraphStore.The addition of the
partialboolean field to the state graph store type is consistent with the broader partial-state tracking introduced across the codebase.
185-197: LGTM: Partial state propagation logic is correct.The logic properly propagates the partial flag through the state graph:
- A delta is partial if the event is partial (
event.isPartial())- OR if the previous delta was partial (
previousDelta?.partial ?? false)This ensures partial state is correctly tracked and persisted through the state chain.
packages/federation-sdk/src/services/staging-area.service.ts (2)
26-26: LGTM: PartialStateResolutionError import.The import is correctly added to support the new error handling path below.
101-106: LGTM: Partial state error handling.The error handling for
PartialStateResolutionErrorcorrectly mirrors the existingMissingAuthorizationEventsErrorpattern, postponing event processing with an appropriate log message when the room is still joining (partial state).packages/federation-sdk/src/services/profiles.service.ts (1)
90-96: LGTM: Early validation prevents invalid join attempts.The validation check ensures the joining user is invited before constructing the membership event. This prevents unnecessary work and provides a clear error message early in the flow.
This aligns well with the broader partial-state handling improvements, as it ensures only valid state transitions are attempted.
packages/core/src/models/event.model.ts (1)
37-37: LGTM: Partial flag added to EventStore.The addition of the
partialboolean field extends the event storage model to track partial state metadata, enabling the persistence and querying of partial events across the codebase.packages/federation-sdk/src/repositories/event.repository.ts (4)
9-9: LGTM: RoomID import added.The import supports the new
findPartialsByRoomIdmethod parameter type below.
391-412: LGTM: Partial parameter added to insertOrUpdateEventWithStateId.The method signature is properly extended with an optional
partialparameter defaulting tofalse, maintaining backward compatibility. The value is correctly persisted in the upsert operation.
433-462: LGTM: Rejection path sets partial: false.The rejection flow correctly sets
partial: falsewhen inserting rejected events, which is appropriate since rejected events should not be considered partial.
475-480: LGTM: Query method for partial events.The new
findPartialsByRoomIdmethod correctly queries events withpartial: truefor a given room, sorted by depth and creation time. The method signature properly uses theRoomIDtype.packages/federation-sdk/src/services/state.service.spec.ts (3)
255-261: LGTM: Helper function for in-memory event store.The
getStorehelper provides a clean abstraction for creating an in-memoryEventStorebacked by aMap, which is useful for testing partial-state scenarios without database dependencies.
263-821: LGTM: Comprehensive partial-state test scenarios.The test data setup creates three distinct partial-state scenarios with realistic event graphs and missing dependencies. This provides good coverage for testing:
- Single missing event in the middle of the chain
- Multiple missing events
- Invite-only room scenarios with missing events
The scenarios are well-structured with proper depth, auth_events, and prev_events chains.
2262-2426: LGTM: Comprehensive partial-state test suite.The new test suite effectively exercises partial-state detection, persistence, and resolution paths:
- Validates incomplete chains are detected
- Confirms partial events are saved and flagged correctly
- Tests partial-state detection at the room level
- Implements a walking algorithm to resolve partial states by fetching and processing missing events
The tests align well with the partial-state functionality introduced across the codebase.
packages/federation-sdk/src/services/event-fetcher.service.ts (1)
90-134: Confirm intentional public access of fetchEventsFromFederation
This method is called from RoomService (packages/federation-sdk/src/services/room.service.ts:1105), so its public visibility is required for cross-service use. If it’s part of the SDK’s public API, add appropriate documentation; otherwise consider marking it internal via JSDoc. No additional security risks beyond existing service boundaries.packages/federation-sdk/src/services/send-join.service.ts (1)
44-47: buildEvent safely handles pre-populated auth_events: it only skips fetching whenauth_eventsis non-empty and will reject any mismatches via the ID sanity check; preservingauth_eventsis required for partial-state support and does not introduce new risks.packages/federation-sdk/src/services/invite.service.ts (2)
54-54: Good change: use roomVersion source of truthSwitch to getRoomVersion(roomId) simplifies and aligns with version-aware factories.
100-137: Room version propagation looks consistentPassing roomVersion to all createFromRawEvent calls keeps event instances consistent across flows.
If any other service still uses getRoomInformation(roomId) only to derive room version, consider switching those to getRoomVersion(roomId) for consistency.
| const servers = await this.stateService.getServerSetInRoom(event.roomId); | ||
|
|
||
| if (event.stateKey) { | ||
| const server = extractDomainFromId(event.stateKey); | ||
| // TODO: fgetser | ||
| if (!servers.has(server)) { | ||
| servers.add(server); | ||
| } | ||
| } |
There was a problem hiding this comment.
Guard domain extraction from state_key; remove stray TODO
extractDomainFromId throws on inputs without a colon (e.g., aliases or third-party invite tokens). Add a safe guard and drop the TODO.
- if (event.stateKey) {
- const server = extractDomainFromId(event.stateKey);
- // TODO: fgetser
- if (!servers.has(server)) {
- servers.add(server);
- }
- }
+ if (event.stateKey?.includes(':')) {
+ const server = extractDomainFromId(event.stateKey);
+ if (!servers.has(server)) {
+ servers.add(server);
+ }
+ } else if (event.stateKey) {
+ this.logger.debug({ stateKey: event.stateKey }, 'state_key has no domain; skipping server augmentation');
+ }🤖 Prompt for AI Agents
In packages/federation-sdk/src/services/federation.service.ts around lines 237
to 245, the call to extractDomainFromId on event.stateKey can throw for inputs
without a colon (aliases or third‑party invite tokens); guard this by either
checking that event.stateKey contains ':' before calling extractDomainFromId or
wrap the call in a try/catch and only add the extracted server to servers if
extraction succeeds, and remove the stray TODO comment.
| try { | ||
| await stateService.getRoomVersion(roomId); | ||
|
|
||
| this.logger.info({ roomId }, 'state already exists'); | ||
| } catch (error) { | ||
| if (error instanceof UnknownRoomError) { | ||
| // if already in room, skip this, walk join event to fill the state | ||
| await stateService.processInitialState(state, authChain); | ||
| } | ||
| } |
There was a problem hiding this comment.
Don’t swallow unexpected errors when checking room existence
Only UnknownRoomError should be handled. Re-throw others to avoid proceeding with an inconsistent state.
- } catch (error) {
- if (error instanceof UnknownRoomError) {
- // if already in room, skip this, walk join event to fill the state
- await stateService.processInitialState(state, authChain);
- }
- }
+ } catch (error) {
+ if (error instanceof UnknownRoomError) {
+ await stateService.processInitialState(state, authChain);
+ } else {
+ throw error;
+ }
+ }🤖 Prompt for AI Agents
In packages/federation-sdk/src/services/room.service.ts around lines 875 to 884,
the try/catch for getRoomVersion currently swallows all errors; modify it so
only UnknownRoomError is handled and all other errors are re-thrown. Concretely,
keep the try to call stateService.getRoomVersion and log on success, catch the
error, check if it is an instance of UnknownRoomError and handle by calling
stateService.processInitialState(state, authChain); otherwise re-throw the
caught error so unexpected failures propagate instead of allowing the method to
continue with inconsistent state.
| const needsFetching = new Set(eventIds) | ||
| .difference(new Set(seenEvents.map((e) => e.eventId))) | ||
| .values() | ||
| .toArray(); | ||
|
|
There was a problem hiding this comment.
🧩 Analysis chain
Nonstandard Set.difference and Iterator.toArray — likely runtime breakage
Set.prototype.difference and iterator .toArray are not widely available unless on very recent runtimes with the new proposals enabled. Replace with portable code.
- const needsFetching = new Set(eventIds)
- .difference(new Set(seenEvents.map((e) => e.eventId)))
- .values()
- .toArray();
+ const seenIds = new Set(seenEvents.map((e) => e.eventId));
+ const needsFetching = eventIds.filter((id) => !seenIds.has(id));🏁 Script executed:
#!/bin/bash
# Verify Node engine target; warn if relying on Set methods / iterator helpers requires Node >= 22
echo "Searching for engines.node declarations..."
fd -t f package.json | while read -r f; do
node_engine=$(jq -r '.engines.node // empty' "$f" 2>/dev/null || true)
if [ -n "$node_engine" ]; then
echo "$f -> engines.node: $node_engine"
fi
done
echo
echo "Grepping for Set.prototype.difference / .toArray usage..."
rg -nP 'Set\\s*\\(.*\\)\\)|\\.difference\\s*\\(|\\.toArray\\s*\\(' -C2 --type=tsLength of output: 176
🏁 Script executed:
#!/bin/bash
echo "Listing engines.node entries in package.json files..."
fd -t f package.json | while read -r f; do
echo "$f -> $(jq -r '.engines.node // "not specified"' "$f")"
done
echo
echo "Searching for .difference( usage in TS files..."
rg -n '\.difference\(' --type ts
echo
echo "Searching for .toArray( usage in TS files..."
rg -n '\.toArray\(' --type tsLength of output: 3820
Avoid unstable Set.difference and Iterator.toArray
Node.js <20 doesn’t support these methods and no engines.node constraint is set. Replace with portable code instead:
- const needsFetching = new Set(eventIds)
- .difference(new Set(seenEvents.map((e) => e.eventId)))
- .values()
- .toArray();
+ const seenIds = new Set(seenEvents.map((e) => e.eventId));
+ const needsFetching = eventIds.filter((id) => !seenIds.has(id));📝 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.
| const needsFetching = new Set(eventIds) | |
| .difference(new Set(seenEvents.map((e) => e.eventId))) | |
| .values() | |
| .toArray(); | |
| const seenIds = new Set(seenEvents.map((e) => e.eventId)); | |
| const needsFetching = eventIds.filter((id) => !seenIds.has(id)); |
🤖 Prompt for AI Agents
In packages/federation-sdk/src/services/room.service.ts around lines 1100 to
1104, the code uses Set.difference and Iterator.toArray which are not supported
on Node.js <20; replace with portable logic: build a Set of seen eventIds (e.g.
new Set(seenEvents.map(e => e.eventId))), then iterate over eventIds (or
Array.from(new Set(eventIds)) to dedupe) and push those not present in the seen
set into an array; assign that array to needsFetching. Ensure no use of
.difference or .toArray so it runs on older Node versions.
There was a problem hiding this comment.
Actionable comments posted: 3
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
Disabled knowledge base sources:
- Jira integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (2)
packages/federation-sdk/src/services/state.service.ts(13 hunks)packages/room/src/manager/factory.ts(3 hunks)
🚧 Files skipped from review as they are similar to previous changes (1)
- packages/room/src/manager/factory.ts
🧰 Additional context used
🧬 Code graph analysis (1)
packages/federation-sdk/src/services/state.service.ts (4)
packages/room/src/types/_common.ts (4)
RoomID(16-16)EventID(8-8)StateID(12-12)StateMapKey(22-22)packages/room/src/manager/factory.ts (1)
PersistentEventFactory(30-123)packages/room/src/types/v3-11.ts (1)
Pdu(736-736)packages/room/src/authorizartion-rules/rules.ts (1)
checkEventAuthWithState(776-888)
🔇 Additional comments (7)
packages/federation-sdk/src/services/state.service.ts (7)
51-62: LGTM: Error classes are well-defined.Both
PartialStateResolutionErrorandUnknownRoomErrorfollow good error handling practices with clear, informative messages.
134-143: LGTM: Partial flag propagation added.The addition of
event.isPartial()to the persistence call correctly propagates partial-state metadata through the storage layer.
196-219: LGTM: Partial awareness in event reconstruction.Correctly passes the
partialflag when reconstructing PDUs from storage, ensuring partial-state metadata is preserved.
544-573: LGTM: Robust partial-state detection.The method correctly handles all three cases (no state, single branch, divergent branches) and properly checks partial flags across deltas.
578-710: LGTM: Partial-state-aware PDU handling.The updates correctly add partial-state checks and early exit paths. The early
PartialStateResolutionError(lines 605-607) appropriately prevents processing when room state is incomplete.Note: The typo in
_neeedsProcessingcall at line 591 is addressed in a separate comment.
777-806: LGTM: Good refactor with backward compatibility.The rename to
getServerSetInRoomreturning aSet<string>is more idiomatic and efficient. The deprecated aliasgetServersInRoommaintains backward compatibility while guiding users to the new API.
1162-1174: LGTM: Partial events retrieval.The new method correctly retrieves and materializes partial events for a room, properly passing the
partialflag during reconstruction.
| async processInitialState(pdus: Pdu[], authChain: Pdu[]) { | ||
| const create = authChain.find((pdu) => pdu.type === 'm.room.create'); | ||
| if (create?.type !== 'm.room.create') { | ||
| throw new Error('No create event found in auth chain to save'); | ||
| } | ||
|
|
||
| const version = create.content.room_version; | ||
|
|
||
| // auth chain for whole state, if sorted by depth, should never have multiples with same branches | ||
| // this confirms correct sorting and being able to save with correct state for each | ||
|
|
||
| // build the map first because .. ?? feels iterative now but makes sense ig | ||
|
|
||
| const authChainCache = new Map<EventID, PersistentEventBase>(); | ||
| for (const pdu of authChain) { | ||
| const event = PersistentEventFactory.createFromRawEvent(pdu, version); | ||
| if (!authChainCache.has(event.eventId)) { | ||
| authChainCache.set(event.eventId, event); | ||
| } | ||
| } | ||
|
|
||
| const eventCache = new Map<EventID, PersistentEventBase>(); | ||
| for (const pdu of pdus) { | ||
| const event = PersistentEventFactory.createFromRawEvent(pdu, version); | ||
| if (eventCache.has(event.eventId) || authChainCache.has(event.eventId)) { | ||
| continue; | ||
| } | ||
| eventCache.set(event.eventId, event); | ||
| } | ||
|
|
||
| // handle create separately | ||
| const createEvent = PersistentEventFactory.createFromRawEvent( | ||
| create, | ||
| version, | ||
| ); | ||
| const stateId = await this.stateRepository.createDelta( | ||
| createEvent, | ||
| '' as StateID, | ||
| ); | ||
| await this.addToRoomGraph(createEvent, stateId); | ||
|
|
||
| this.logger.info( | ||
| { eventId: createEvent.eventId, roomId: createEvent.roomId, stateId }, | ||
| 'create event saved', | ||
| ); | ||
|
|
||
| const getAuthEventStateMap = (e: PersistentEventBase) => { | ||
| return e.getAuthEventIds().reduce((accum, curr) => { | ||
| // every event should have it's auth events in the auth chain | ||
| const event = authChainCache.get(curr); | ||
| if (event) { | ||
| accum.set(event.getUniqueStateIdentifier(), event); | ||
| } | ||
| return accum; | ||
| }, new Map<StateMapKey, PersistentEventBase>()); | ||
| }; | ||
|
|
||
| const store = this._getStore(version); | ||
|
|
||
| const sortedEvents = Array.from(eventCache.values()) | ||
| .concat(Array.from(authChainCache.values())) | ||
| .sort((e1, e2) => { | ||
| if (e1.depth !== e2.depth) { | ||
| return e1.depth - e2.depth; | ||
| } | ||
|
|
||
| if (e1.originServerTs !== e2.originServerTs) { | ||
| return e1.originServerTs - e2.originServerTs; | ||
| } | ||
|
|
||
| return e1.eventId.localeCompare(e2.eventId); | ||
| }); | ||
|
|
||
| let previousStateId = stateId; | ||
|
|
||
| for (const event of sortedEvents) { | ||
| const authState = getAuthEventStateMap(event); | ||
| try { | ||
| await checkEventAuthWithState(event, authState, store); | ||
| } catch (error) { | ||
| this.logger.error( | ||
| { | ||
| eventId: event.eventId, | ||
| authEvents: event.getAuthEventIds(), | ||
| }, | ||
| 'event failed auth check while saving state, this should not have happened while walking an auth chain, the chain is incorrect', | ||
| ); | ||
|
|
||
| // propagating throw, at this point this is not supposed to fail, something is wrong with the state we received | ||
| throw error; | ||
| } | ||
|
|
||
| // auth events themseleves can be partial at any point | ||
| event.setPartial( | ||
| // if some of the previous events are partial this one also needs to be partial | ||
| event | ||
| .getPreviousEventIds() | ||
| .some((id) => { | ||
| const event = authChainCache.get(id) || eventCache.get(id); | ||
| // event notseen | ||
| if (!event) { | ||
| return true; | ||
| } | ||
|
|
||
| // seen event is also partial | ||
| return event.isPartial(); | ||
| }), | ||
| ); | ||
| previousStateId = await this.stateRepository.createDelta( | ||
| event, | ||
| previousStateId, | ||
| ); | ||
| await this.addToRoomGraph(event, previousStateId); | ||
| } | ||
|
|
||
| return previousStateId; | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
Refactor: Extract helpers to reduce complexity.
The processInitialState method is 117 lines long and handles multiple responsibilities (caching, sorting, auth checking, partial-state propagation, persistence). This makes it difficult to test, maintain, and reason about.
Consider extracting these helper methods:
_buildEventCaches(pdus, authChain, version)– Build and return auth chain and event caches_sortEventsByDepthAndTime(events)– Sort events by depth, timestamp, and event ID_determineIfPartial(event, authChainCache, eventCache)– Extract lines 486-500 into a dedicated method_persistEventInChain(event, authState, previousStateId, store, version)– Handle auth check and persistence
Example structure:
async processInitialState(pdus: Pdu[], authChain: Pdu[]) {
const create = authChain.find((pdu) => pdu.type === 'm.room.create');
if (create?.type !== 'm.room.create') {
throw new Error('No create event found in auth chain to save');
}
const version = create.content.room_version;
- const authChainCache = new Map<EventID, PersistentEventBase>();
- for (const pdu of authChain) {
- const event = PersistentEventFactory.createFromRawEvent(pdu, version);
- if (!authChainCache.has(event.eventId)) {
- authChainCache.set(event.eventId, event);
- }
- }
-
- const eventCache = new Map<EventID, PersistentEventBase>();
- for (const pdu of pdus) {
- const event = PersistentEventFactory.createFromRawEvent(pdu, version);
- if (eventCache.has(event.eventId) || authChainCache.has(event.eventId)) {
- continue;
- }
- eventCache.set(event.eventId, event);
- }
+ const { authChainCache, eventCache } = this._buildEventCaches(pdus, authChain, version);
// handle create separately...
const createEvent = PersistentEventFactory.createFromRawEvent(create, version);
const stateId = await this.stateRepository.createDelta(createEvent, '' as StateID);
await this.addToRoomGraph(createEvent, stateId);
- const getAuthEventStateMap = (e: PersistentEventBase) => { ... };
const store = this._getStore(version);
- const sortedEvents = Array.from(eventCache.values())
- .concat(Array.from(authChainCache.values()))
- .sort((e1, e2) => { ... });
+ const sortedEvents = this._sortEventsByDepthAndTime([...eventCache.values(), ...authChainCache.values()]);
let previousStateId = stateId;
for (const event of sortedEvents) {
- const authState = getAuthEventStateMap(event);
+ const authState = this._getAuthEventStateMap(event, authChainCache);
try {
await checkEventAuthWithState(event, authState, store);
} catch (error) {
// ... error handling
throw error;
}
- event.setPartial(
- event.getPreviousEventIds().some((id) => {
- const event = authChainCache.get(id) || eventCache.get(id);
- if (!event) return true;
- return event.isPartial();
- }),
- );
+ event.setPartial(this._determineIfPartial(event, authChainCache, eventCache));
+
previousStateId = await this.stateRepository.createDelta(event, previousStateId);
await this.addToRoomGraph(event, previousStateId);
}
return previousStateId;
}
+
+private _buildEventCaches(pdus: Pdu[], authChain: Pdu[], version: RoomVersion) {
+ const authChainCache = new Map<EventID, PersistentEventBase>();
+ for (const pdu of authChain) {
+ const event = PersistentEventFactory.createFromRawEvent(pdu, version);
+ if (!authChainCache.has(event.eventId)) {
+ authChainCache.set(event.eventId, event);
+ }
+ }
+
+ const eventCache = new Map<EventID, PersistentEventBase>();
+ for (const pdu of pdus) {
+ const event = PersistentEventFactory.createFromRawEvent(pdu, version);
+ if (eventCache.has(event.eventId) || authChainCache.has(event.eventId)) {
+ continue;
+ }
+ eventCache.set(event.eventId, event);
+ }
+
+ return { authChainCache, eventCache };
+}
+
+private _sortEventsByDepthAndTime(events: PersistentEventBase[]) {
+ return events.sort((e1, e2) => {
+ if (e1.depth !== e2.depth) return e1.depth - e2.depth;
+ if (e1.originServerTs !== e2.originServerTs) return e1.originServerTs - e2.originServerTs;
+ return e1.eventId.localeCompare(e2.eventId);
+ });
+}
+
+private _determineIfPartial(
+ event: PersistentEventBase,
+ authChainCache: Map<EventID, PersistentEventBase>,
+ eventCache: Map<EventID, PersistentEventBase>,
+): boolean {
+ return event.getPreviousEventIds().some((id) => {
+ const prevEvent = authChainCache.get(id) || eventCache.get(id);
+ return !prevEvent || prevEvent.isPartial();
+ });
+}
+
+private _getAuthEventStateMap(
+ event: PersistentEventBase,
+ authChainCache: Map<EventID, PersistentEventBase>,
+) {
+ return event.getAuthEventIds().reduce((accum, curr) => {
+ const authEvent = authChainCache.get(curr);
+ if (authEvent) {
+ accum.set(authEvent.getUniqueStateIdentifier(), authEvent);
+ }
+ return accum;
+ }, new Map<StateMapKey, PersistentEventBase>());
+}This refactor improves testability (each helper can be unit tested), readability (clear single-purpose methods), and maintainability (easier to modify individual steps).
| private async _neeedsProcessing<P extends PersistentEventBase>( | ||
| event: P, | ||
| ): Promise<P | null> { | ||
| const record = await this.eventRepository.findById(event.eventId); | ||
| if (record?.partial) { | ||
| // event is saved and is partial, pass it | ||
| event.setPartial(true); | ||
| return event; | ||
| } | ||
|
|
||
| const previousEvents = await this.eventRepository | ||
| .findByIds(event.getPreviousEventIds()) | ||
| .toArray(); | ||
| if (previousEvents.length !== event.getPreviousEventIds().length) { | ||
| // if we don't have all the previous events now, this is a partial state | ||
| event.setPartial(true); | ||
| return event; | ||
| } | ||
|
|
||
| if (previousEvents.some((e) => e.partial)) { | ||
| // if any of the previouseventsis partial this is too | ||
| event.setPartial(true); | ||
| return event; | ||
| } | ||
|
|
||
| // isn't partial, check if already stored, then skip | ||
| if (record) { | ||
| return null; | ||
| } | ||
|
|
||
| return event; | ||
| } |
There was a problem hiding this comment.
Fix typo in method name.
The private method name _neeedsProcessing has three 'e's and should be _needsProcessing.
Apply this diff:
-private async _neeedsProcessing<P extends PersistentEventBase>(
+private async _needsProcessing<P extends PersistentEventBase>(
event: P,
): Promise<P | null> {Also update the call site at line 591:
-const event = await this._neeedsProcessing(pdu);
+const event = await this._needsProcessing(pdu);📝 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.
| private async _neeedsProcessing<P extends PersistentEventBase>( | |
| event: P, | |
| ): Promise<P | null> { | |
| const record = await this.eventRepository.findById(event.eventId); | |
| if (record?.partial) { | |
| // event is saved and is partial, pass it | |
| event.setPartial(true); | |
| return event; | |
| } | |
| const previousEvents = await this.eventRepository | |
| .findByIds(event.getPreviousEventIds()) | |
| .toArray(); | |
| if (previousEvents.length !== event.getPreviousEventIds().length) { | |
| // if we don't have all the previous events now, this is a partial state | |
| event.setPartial(true); | |
| return event; | |
| } | |
| if (previousEvents.some((e) => e.partial)) { | |
| // if any of the previouseventsis partial this is too | |
| event.setPartial(true); | |
| return event; | |
| } | |
| // isn't partial, check if already stored, then skip | |
| if (record) { | |
| return null; | |
| } | |
| return event; | |
| } | |
| private async _needsProcessing<P extends PersistentEventBase>( | |
| event: P, | |
| ): Promise<P | null> { | |
| const record = await this.eventRepository.findById(event.eventId); | |
| if (record?.partial) { | |
| // event is saved and is partial, pass it | |
| event.setPartial(true); | |
| return event; | |
| } | |
| const previousEvents = await this.eventRepository | |
| .findByIds(event.getPreviousEventIds()) | |
| .toArray(); | |
| if (previousEvents.length !== event.getPreviousEventIds().length) { | |
| // if we don't have all the previous events now, this is a partial state | |
| event.setPartial(true); | |
| return event; | |
| } | |
| if (previousEvents.some((e) => e.partial)) { | |
| // if any of the previous events is partial this is too | |
| event.setPartial(true); | |
| return event; | |
| } | |
| // isn't partial, check if already stored, then skip | |
| if (record) { | |
| return null; | |
| } | |
| return event; | |
| } |
| private async _neeedsProcessing<P extends PersistentEventBase>( | |
| event: P, | |
| ): Promise<P | null> { | |
| const record = await this.eventRepository.findById(event.eventId); | |
| if (record?.partial) { | |
| // event is saved and is partial, pass it | |
| event.setPartial(true); | |
| return event; | |
| } | |
| const previousEvents = await this.eventRepository | |
| .findByIds(event.getPreviousEventIds()) | |
| .toArray(); | |
| if (previousEvents.length !== event.getPreviousEventIds().length) { | |
| // if we don't have all the previous events now, this is a partial state | |
| event.setPartial(true); | |
| return event; | |
| } | |
| if (previousEvents.some((e) => e.partial)) { | |
| // if any of the previouseventsis partial this is too | |
| event.setPartial(true); | |
| return event; | |
| } | |
| // isn't partial, check if already stored, then skip | |
| if (record) { | |
| return null; | |
| } | |
| return event; | |
| } | |
| const event = await this._needsProcessing(pdu); |
🤖 Prompt for AI Agents
In packages/federation-sdk/src/services/state.service.ts around lines 511 to
542, the private method name `_neeedsProcessing` is misspelled with an extra 'e'
and should be renamed to `_needsProcessing`; rename the method declaration and
all internal references, and update its call site at line 591 to use
`_needsProcessing` so the symbol matches. Ensure exports/usage within the class
remain consistent and run a project-wide search to update any other references
to the old name.
| throw new Error(`no previous state for event ${event.eventId}`); | ||
| } | ||
|
|
||
| if (event.isPartial()) { | ||
| // walked over to this, since we have the state at this event, toggle event to be not partial any longer | ||
| this.logger.debug( | ||
| { eventId: event.eventId }, | ||
| 'completing state at event', | ||
| ); | ||
| // previous states by this point should NOT be partial | ||
| event.setPartial(!event.isPartial()); |
There was a problem hiding this comment.
Simplify confusing double-negative logic.
Line 946 uses event.setPartial(!event.isPartial()) which is unnecessarily confusing. Since the comment states "completing state at event" and the condition checks if (event.isPartial()), the intent is clear: mark the event as no longer partial.
Apply this diff to make the logic explicit:
if (event.isPartial()) {
- // walked over to this, since we have the state at this event, toggle event to be not partial any longer
+ // walked over to this, since we have the state at this event, mark event as no longer partial
this.logger.debug(
{ eventId: event.eventId },
'completing state at event',
);
- // previous states by this point should NOT be partial
- event.setPartial(!event.isPartial());
+ event.setPartial(false);
}This removes the double negative and makes the intent crystal clear: when we have complete state, explicitly set partial to false.
📝 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.
| throw new Error(`no previous state for event ${event.eventId}`); | |
| } | |
| if (event.isPartial()) { | |
| // walked over to this, since we have the state at this event, toggle event to be not partial any longer | |
| this.logger.debug( | |
| { eventId: event.eventId }, | |
| 'completing state at event', | |
| ); | |
| // previous states by this point should NOT be partial | |
| event.setPartial(!event.isPartial()); | |
| throw new Error(`no previous state for event ${event.eventId}`); | |
| } | |
| if (event.isPartial()) { | |
| // walked over to this, since we have the state at this event, mark event as no longer partial | |
| this.logger.debug( | |
| { eventId: event.eventId }, | |
| 'completing state at event', | |
| ); | |
| event.setPartial(false); | |
| } |
🤖 Prompt for AI Agents
In packages/federation-sdk/src/services/state.service.ts around lines 936 to
946, the code uses a confusing double-negative by calling
event.setPartial(!event.isPartial()) inside an if (event.isPartial()) block;
replace that with an explicit event.setPartial(false) to clearly mark the event
as no longer partial and keep the comment and debug log intact.
Summary by CodeRabbit