refactor: improve replication lag reporting accuracy and optimize Raf…#28
Conversation
…t log compaction logic
📝 WalkthroughWalkthroughThe PR updates broker Raft election, snapshot, compaction, and telemetry behavior, raises the default compaction threshold, and expands the TypeScript client with browser-compatible binary types and serialized error codes. Documentation and minor client cleanup are also included. ChangesBroker Raft behavior
TypeScript client compatibility
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Scheduler
participant RaftNode
participant RaftLog
participant Peer
Scheduler->>RaftNode: run periodic Raft task
RaftNode->>RaftLog: inspect replication and compaction state
RaftNode->>Peer: send snapshot when peer log is behind
Peer-->>RaftNode: provide snapshot chunks
RaftNode->>RaftLog: track snapshot and compact committed entries
Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
drmq-ts-client/src/client.ts (1)
250-252: 🩺 Stability & Availability | 🔴 Critical | ⚡ Quick winRemove
Bufferusage to maintain browser compatibility.The PR objective introduces browser-compatible binary handling by replacing
BufferwithUint8Arrayinmessages.ts. However,Buffer.fromis still being used here. In a browser environment without aBufferpolyfill, this will throw aReferenceErrorand crash the client.Since
pm.payloadis already securely copied and initialized as aUint8Arrayin the upstreamsend()method, and the protobuf messages now natively acceptUint8Array, you can assign it directly.🐛 Proposed fix
entries: batch.map(pm => ({ - payload: Buffer.from(pm.payload), + payload: pm.payload, key: pm.key, clientTimestamp: pm.timestamp🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@drmq-ts-client/src/client.ts` around lines 250 - 252, Replace the Buffer.from conversion in the batch entries mapping with direct assignment of pm.payload. Preserve the existing payload and key mapping while relying on the Uint8Array initialized by send() for browser-compatible protobuf handling.drmq-broker/src/main/java/com/drmq/broker/raft/RaftNode.java (1)
422-437: 🩺 Stability & Availability | 🔴 Critical | ⚡ Quick winHandle the zero-peer case in
startElection
votesReceivedstarts at 1, but quorum is only checked in the per-peer callback. With an emptypeerslist, that callback never runs, so a standalone node never reachesbecomeLeader().🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@drmq-broker/src/main/java/com/drmq/broker/raft/RaftNode.java` around lines 422 - 437, The startElection method must handle an empty peers list because no per-peer callback will evaluate the self-vote. After initializing votesNeeded, votesReceived, and electionWon, immediately call becomeLeader() when the node’s self-vote satisfies quorum for a zero-peer cluster, while preserving the existing peer-election callback flow otherwise.
🧹 Nitpick comments (1)
drmq-broker/src/main/java/com/drmq/broker/TelemetryWebSocketServer.java (1)
308-338: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winBounded but repeated synchronized Raft-log scans on the telemetry thread.
For each peer with
lagEntriesin(0, 2000], this now calls thesynchronizedRaftLog.getEntry(idx)(drmq-broker/src/main/java/com/drmq/broker/raft/RaftLog.java, lines 138-143) up to 2000 times, plus a protobufparseFromperBATCH_MESSAGEentry, once per second per peer (onOpentriggers it too, per new dashboard connection).RaftLog's synchronized monitor is shared with the leader's real append/compact hot path, so this adds lock contention exactly when replication lag (and thus system load) is already elevated — the scenario telemetry should avoid perturbing.Consider capping the exact-scan threshold well below 2000, or maintaining an incrementally-updated message-count counter on the Raft log/entry-apply path instead of re-scanning and re-parsing entries on every broadcast tick.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@drmq-broker/src/main/java/com/drmq/broker/TelemetryWebSocketServer.java` around lines 308 - 338, Reduce or remove the per-peer exact Raft-log scan in the replication-lag calculation around the telemetry broadcast logic. Prefer a substantially lower lagEntries threshold with the existing estimate fallback, or reuse an incrementally maintained message-count metric from the Raft log/apply path; do not repeatedly call RaftLog.getEntry or parse BATCH_MESSAGE entries for up to 2000 entries on the telemetry thread.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@drmq-broker/src/main/java/com/drmq/broker/raft/RaftNode.java`:
- Around line 422-437: The startElection method must handle an empty peers list
because no per-peer callback will evaluate the self-vote. After initializing
votesNeeded, votesReceived, and electionWon, immediately call becomeLeader()
when the node’s self-vote satisfies quorum for a zero-peer cluster, while
preserving the existing peer-election callback flow otherwise.
In `@drmq-ts-client/src/client.ts`:
- Around line 250-252: Replace the Buffer.from conversion in the batch entries
mapping with direct assignment of pm.payload. Preserve the existing payload and
key mapping while relying on the Uint8Array initialized by send() for
browser-compatible protobuf handling.
---
Nitpick comments:
In `@drmq-broker/src/main/java/com/drmq/broker/TelemetryWebSocketServer.java`:
- Around line 308-338: Reduce or remove the per-peer exact Raft-log scan in the
replication-lag calculation around the telemetry broadcast logic. Prefer a
substantially lower lagEntries threshold with the existing estimate fallback, or
reuse an incrementally maintained message-count metric from the Raft log/apply
path; do not repeatedly call RaftLog.getEntry or parse BATCH_MESSAGE entries for
up to 2000 entries on the telemetry thread.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 18f933b8-8a81-43fd-b16c-d5816d65e347
📒 Files selected for processing (8)
README.mddrmq-broker/src/main/java/com/drmq/broker/BrokerConfig.javadrmq-broker/src/main/java/com/drmq/broker/TelemetryWebSocketServer.javadrmq-broker/src/main/java/com/drmq/broker/raft/RaftNode.javadrmq-client/src/main/java/com/drmq/client/DRMQConsumer.javadrmq-ts-client/src/client.tsdrmq-ts-client/src/example.tsdrmq-ts-client/src/messages.ts
💤 Files with no reviewable changes (1)
- drmq-ts-client/src/example.ts
1 participant
…t log compaction logic
Summary by CodeRabbit
New Features
Bug Fixes
Documentation