feat(epaxos): retire executed residents beyond retention - #5
Conversation
Reviewer's GuideReplaces the legacy conflict-index maps with a radix-tree-based conflictEngine, adds executed-instance retirement with an async record-load handshake, tightens validation and error handling across EPaxos core, tests, and KV examples, and wires new configuration knobs and metrics into the node and fault campaigns. Sequence diagram for the new Ready.RecordLoads / ProvideRecordLoad handshakesequenceDiagram
actor Client
participant RawNode
participant Storage
Client->>RawNode: Step(Message)
activate RawNode
RawNode->>RawNode: needsRecordLoad(ref)
alt record load required
RawNode->>RawNode: deferRecordLoad(ref, Message)
RawNode-->>Client: error ErrMessageRejected (if queue full)
deactivate RawNode
Storage->>RawNode: Ready()
note over RawNode: Ready.RecordLoads contains ref
RawNode-->>Storage: Ready{RecordLoads}
Storage->>Storage: load InstanceRecord from durable
Storage->>RawNode: ProvideRecordLoad(RecordLoadResult)
activate RawNode
RawNode->>RawNode: validateRecordChecksum(rec)
RawNode->>RawNode: installInstance / setInstanceRecord
RawNode->>RawNode: Step(deferred Message)
RawNode->>RawNode: maybeRefoldLoaded(ref)
deactivate RawNode
else no record load required
RawNode->>RawNode: handlePrepare / handleEvidence
RawNode-->>Client: Step result
end
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
|
Closes #12 |
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 2 issues
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="epaxos/retire_test.go" line_range="8" />
<code_context>
+ "testing"
+)
+
+func TestExecutedTrackerContainsFoldedPrefix(t *testing.T) {
+ var tr executedTracker
+ lane := instanceLane{conf: 1, replica: 1}
</code_context>
<issue_to_address>
**suggestion (testing):** Consider adding tests for retainExecutedPerLane edge cases, especially zero retention
Right now the tests only cover positive RetainExecutedPerLane. Please also add a case with RetainExecutedPerLane = 0 that verifies:
- all executed instances on a lane can be folded, and
- conflict engine foldedThrough watermark and executedTracker.contains remain consistent.
It would also be useful to cover a scenario where the number of executed instances is smaller than the retention window.
Suggested implementation:
```golang
if !tr.contains(InstanceRef{Conf: 1, Replica: 1, Instance: 2}) {
t.Fatal("expected contains via through after exact cleanup")
}
}
func TestExecutedTrackerZeroRetentionFoldsAll(t *testing.T) {
var tr executedTracker
lane := instanceLane{conf: 1, replica: 1}
// Simulate executing a contiguous range of instances on a lane.
for i := InstanceNum(1); i <= 5; i++ {
tr.add(InstanceRef{Conf: 1, Replica: 1, Instance: i})
}
// With RetainExecutedPerLane = 0, all executed instances on this lane
// should be eligible to be folded, i.e. the engine would be allowed
// to forget the entire prefix.
//
// We simulate that by folding the prefix completely via forgetExactThrough.
tr.forgetExactThrough(lane, 5)
// The foldedThrough watermark (queried via prefix) should reflect that
// all executed instances up to 5 have been folded.
if got := tr.prefix(lane); got != 5 {
t.Fatalf("prefix after zero-retention fold = %d, want 5", got)
}
// Even though everything is folded, contains() must still report all
// folded instances as executed, using the foldedThrough watermark.
for i := InstanceNum(1); i <= 5; i++ {
if !tr.contains(InstanceRef{Conf: 1, Replica: 1, Instance: i}) {
t.Fatalf("expected contains(%d) to be true after zero-retention fold", i)
}
}
}
func TestExecutedTrackerRetentionWindowLargerThanExecuted(t *testing.T) {
var tr executedTracker
lane := instanceLane{conf: 1, replica: 1}
// Simulate executing fewer instances than a hypothetical retention window.
// For example, if RetainExecutedPerLane = 10 and we only executed 3
// instances, the engine should not need to fold anything yet.
for i := InstanceNum(1); i <= 3; i++ {
tr.add(InstanceRef{Conf: 1, Replica: 1, Instance: i})
}
// No folding is applied here, to reflect that the retention window is
// larger than the executed range.
// All instances should be contained and the prefix should point at the
// highest executed instance.
if got := tr.prefix(lane); got != 3 {
t.Fatalf("prefix with small executed set and large retention window = %d, want 3", got)
}
for i := InstanceNum(1); i <= 3; i++ {
if !tr.contains(InstanceRef{Conf: 1, Replica: 1, Instance: i}) {
t.Fatalf("expected contains(%d) to be true when retention window > executed count", i)
}
}
}
```
To fully exercise `retainExecutedPerLane` and the conflict engine’s `foldedThrough` watermark as mentioned in your review comment, you will likely want to:
1. Add an integration-style test (probably in this same file) that:
- Constructs a real conflict engine instance with `RetainExecutedPerLane = 0`.
- Marks a contiguous sequence of instances as executed through the engine’s public API.
- Triggers whatever mechanism applies retention (e.g. a `tick`, `garbageCollect`, or similar method that folds executed instances based on `RetainExecutedPerLane`).
- Asserts that:
* The engine’s `foldedThrough` watermark for the lane equals the highest executed instance.
* The internal `executedTracker` (reachable via exported accessors or test hooks) reports `contains()` correctly for folded instances.
2. Add a parallel test where `RetainExecutedPerLane` is set to a value greater than the number of executed instances (e.g. `RetainExecutedPerLane = 10` with only 3 executed instances), and verify that:
- No folding occurs (the `foldedThrough` watermark does not advance unexpectedly).
- All executed instances are still individually tracked (if the implementation distinguishes between folded and explicitly tracked).
You will need to adapt the above tests to the actual conflict engine type and API present in your codebase (constructor, configuration struct/fields, and the method used to drive retention).
</issue_to_address>
### Comment 2
<location path="epaxos/record_load_test.go" line_range="25" />
<code_context>
+ engine.advanceFold(lane, rec.Ref.Instance)
+}
+
+func TestRecordLoadFoundReplaysDeferred(t *testing.T) {
+ rn, err := NewRawNode(Config{ID: 1, Voters: makeIDs(3)})
+ if err != nil {
</code_context>
<issue_to_address>
**suggestion (testing):** Add tests for ProvideRecordLoad Ref handling (zero Ref override and mismatched Ref error case)
One behavior that's not covered is how ProvideRecordLoad handles the Ref field:
- If Record.Ref is zero and the result Ref is non-zero, it updates rec.Ref = res.Ref.
- If Record.Ref is non-zero and differs from res.Ref, it should return an error.
Please add tests for these two cases to lock in the expected override behavior and ensure we fail when a record would be installed under the wrong InstanceRef.
Suggested implementation:
```golang
func TestRecordLoadFoundReplaysDeferred(t *testing.T) {
rn, err := NewRawNode(Config{ID: 1, Voters: makeIDs(3)})
if err != nil {
t.Fatal(err)
}
ref := InstanceRef{Conf: 1, Replica: 2, Instance: 5}
rec := checkedRecord(InstanceRecord{
Ref: ref, Status: StatusCommitted, Seq: 3, Ballot: Ballot{Replica: 2},
Deps: rn.q.deps(), Command: Command{Payload: []byte("x"), ConflictKeys: [][]byte{[]byte("k")}},
})
// Mark folded without resident.
foldTestRef(&rn.engine, rec)
}
func TestProvideRecordLoadZeroRefOverride(t *testing.T) {
rn, err := NewRawNode(Config{ID: 1, Voters: makeIDs(3)})
if err != nil {
t.Fatal(err)
}
// Create a resident record that will be "loaded" by ProvideRecordLoad.
ref := InstanceRef{Conf: 1, Replica: 2, Instance: 5}
stored := checkedRecord(InstanceRecord{
Ref: ref,
Status: StatusCommitted,
Seq: 3,
Ballot: Ballot{Replica: 2},
Deps: rn.q.deps(),
Command: Command{
Payload: []byte("x"),
ConflictKeys: [][]byte{[]byte("k")},
},
})
// Install the resident record in the engine so ProvideRecordLoad can find it.
foldTestRef(&rn.engine, stored)
// Build a load request with zero Ref that should be overridden by the result Ref.
rec := stored
rec.Ref = InstanceRef{} // zero Ref triggers override behavior
lane := 0 // use the default lane for this test
loaded, err := rn.engine.ProvideRecordLoad(lane, rec)
if err != nil {
t.Fatalf("ProvideRecordLoad unexpected error: %v", err)
}
if loaded.Ref != ref {
t.Fatalf("expected loaded.Ref to be %v, got %v", ref, loaded.Ref)
}
// Verify that the input record Ref was also updated to match the loaded Ref.
if rec.Ref != ref {
t.Fatalf("expected rec.Ref to be updated to %v, got %v", ref, rec.Ref)
}
}
func TestProvideRecordLoadMismatchedRefError(t *testing.T) {
rn, err := NewRawNode(Config{ID: 1, Voters: makeIDs(3)})
if err != nil {
t.Fatal(err)
}
// Create a resident record with a specific Ref.
ref := InstanceRef{Conf: 1, Replica: 2, Instance: 5}
stored := checkedRecord(InstanceRecord{
Ref: ref,
Status: StatusCommitted,
Seq: 3,
Ballot: Ballot{Replica: 2},
Deps: rn.q.deps(),
Command: Command{
Payload: []byte("x"),
ConflictKeys: [][]byte{[]byte("k")},
},
})
// Install the resident record in the engine so ProvideRecordLoad can find it.
foldTestRef(&rn.engine, stored)
// Build a load request whose Ref does not match the stored Ref.
rec := stored
rec.Ref = InstanceRef{Conf: 9, Replica: 9, Instance: 9}
lane := 0
_, err = rn.engine.ProvideRecordLoad(lane, rec)
if err == nil {
t.Fatalf("expected ProvideRecordLoad to fail for mismatched Ref, got nil error")
}
}
```
1. These tests assume `ProvideRecordLoad` is a method with signature `func (e *engine) ProvideRecordLoad(lane int, rec InstanceRecord) (InstanceRecord, error)`. If the actual receiver type, lane type, parameter passing (by value vs pointer), or return values differ, adjust the call sites accordingly.
2. If `foldTestRef` is not sufficient to make the record discoverable by `ProvideRecordLoad`, you may need to mirror whatever setup other `ProvideRecordLoad` tests use (e.g., calling specific engine helpers to install the resident record).
3. If `InstanceRef` comparison requires a helper (e.g., `EqualInstanceRef`) instead of `!=`/`==`, replace the direct comparisons with that helper to stay consistent with existing conventions.
4. If `record_load_test.go` does not yet import any additional packages (like `reflect`) and you choose to use them for comparisons, add the imports at the top of the file.
5. If `TestRecordLoadFoundReplaysDeferred` contains additional code after the snippet you provided, move the newly added tests so they are defined at file scope (i.e., after the closing brace of that test) to keep the file compiling.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| "testing" | ||
| ) | ||
|
|
||
| func TestExecutedTrackerContainsFoldedPrefix(t *testing.T) { |
There was a problem hiding this comment.
suggestion (testing): Consider adding tests for retainExecutedPerLane edge cases, especially zero retention
Right now the tests only cover positive RetainExecutedPerLane. Please also add a case with RetainExecutedPerLane = 0 that verifies:
- all executed instances on a lane can be folded, and
- conflict engine foldedThrough watermark and executedTracker.contains remain consistent.
It would also be useful to cover a scenario where the number of executed instances is smaller than the retention window.
Suggested implementation:
if !tr.contains(InstanceRef{Conf: 1, Replica: 1, Instance: 2}) {
t.Fatal("expected contains via through after exact cleanup")
}
}
func TestExecutedTrackerZeroRetentionFoldsAll(t *testing.T) {
var tr executedTracker
lane := instanceLane{conf: 1, replica: 1}
// Simulate executing a contiguous range of instances on a lane.
for i := InstanceNum(1); i <= 5; i++ {
tr.add(InstanceRef{Conf: 1, Replica: 1, Instance: i})
}
// With RetainExecutedPerLane = 0, all executed instances on this lane
// should be eligible to be folded, i.e. the engine would be allowed
// to forget the entire prefix.
//
// We simulate that by folding the prefix completely via forgetExactThrough.
tr.forgetExactThrough(lane, 5)
// The foldedThrough watermark (queried via prefix) should reflect that
// all executed instances up to 5 have been folded.
if got := tr.prefix(lane); got != 5 {
t.Fatalf("prefix after zero-retention fold = %d, want 5", got)
}
// Even though everything is folded, contains() must still report all
// folded instances as executed, using the foldedThrough watermark.
for i := InstanceNum(1); i <= 5; i++ {
if !tr.contains(InstanceRef{Conf: 1, Replica: 1, Instance: i}) {
t.Fatalf("expected contains(%d) to be true after zero-retention fold", i)
}
}
}
func TestExecutedTrackerRetentionWindowLargerThanExecuted(t *testing.T) {
var tr executedTracker
lane := instanceLane{conf: 1, replica: 1}
// Simulate executing fewer instances than a hypothetical retention window.
// For example, if RetainExecutedPerLane = 10 and we only executed 3
// instances, the engine should not need to fold anything yet.
for i := InstanceNum(1); i <= 3; i++ {
tr.add(InstanceRef{Conf: 1, Replica: 1, Instance: i})
}
// No folding is applied here, to reflect that the retention window is
// larger than the executed range.
// All instances should be contained and the prefix should point at the
// highest executed instance.
if got := tr.prefix(lane); got != 3 {
t.Fatalf("prefix with small executed set and large retention window = %d, want 3", got)
}
for i := InstanceNum(1); i <= 3; i++ {
if !tr.contains(InstanceRef{Conf: 1, Replica: 1, Instance: i}) {
t.Fatalf("expected contains(%d) to be true when retention window > executed count", i)
}
}
}To fully exercise retainExecutedPerLane and the conflict engine’s foldedThrough watermark as mentioned in your review comment, you will likely want to:
-
Add an integration-style test (probably in this same file) that:
- Constructs a real conflict engine instance with
RetainExecutedPerLane = 0. - Marks a contiguous sequence of instances as executed through the engine’s public API.
- Triggers whatever mechanism applies retention (e.g. a
tick,garbageCollect, or similar method that folds executed instances based onRetainExecutedPerLane). - Asserts that:
- The engine’s
foldedThroughwatermark for the lane equals the highest executed instance. - The internal
executedTracker(reachable via exported accessors or test hooks) reportscontains()correctly for folded instances.
- The engine’s
- Constructs a real conflict engine instance with
-
Add a parallel test where
RetainExecutedPerLaneis set to a value greater than the number of executed instances (e.g.RetainExecutedPerLane = 10with only 3 executed instances), and verify that:- No folding occurs (the
foldedThroughwatermark does not advance unexpectedly). - All executed instances are still individually tracked (if the implementation distinguishes between folded and explicitly tracked).
- No folding occurs (the
You will need to adapt the above tests to the actual conflict engine type and API present in your codebase (constructor, configuration struct/fields, and the method used to drive retention).
| engine.advanceFold(lane, rec.Ref.Instance) | ||
| } | ||
|
|
||
| func TestRecordLoadFoundReplaysDeferred(t *testing.T) { |
There was a problem hiding this comment.
suggestion (testing): Add tests for ProvideRecordLoad Ref handling (zero Ref override and mismatched Ref error case)
One behavior that's not covered is how ProvideRecordLoad handles the Ref field:
- If Record.Ref is zero and the result Ref is non-zero, it updates rec.Ref = res.Ref.
- If Record.Ref is non-zero and differs from res.Ref, it should return an error.
Please add tests for these two cases to lock in the expected override behavior and ensure we fail when a record would be installed under the wrong InstanceRef.
Suggested implementation:
func TestRecordLoadFoundReplaysDeferred(t *testing.T) {
rn, err := NewRawNode(Config{ID: 1, Voters: makeIDs(3)})
if err != nil {
t.Fatal(err)
}
ref := InstanceRef{Conf: 1, Replica: 2, Instance: 5}
rec := checkedRecord(InstanceRecord{
Ref: ref, Status: StatusCommitted, Seq: 3, Ballot: Ballot{Replica: 2},
Deps: rn.q.deps(), Command: Command{Payload: []byte("x"), ConflictKeys: [][]byte{[]byte("k")}},
})
// Mark folded without resident.
foldTestRef(&rn.engine, rec)
}
func TestProvideRecordLoadZeroRefOverride(t *testing.T) {
rn, err := NewRawNode(Config{ID: 1, Voters: makeIDs(3)})
if err != nil {
t.Fatal(err)
}
// Create a resident record that will be "loaded" by ProvideRecordLoad.
ref := InstanceRef{Conf: 1, Replica: 2, Instance: 5}
stored := checkedRecord(InstanceRecord{
Ref: ref,
Status: StatusCommitted,
Seq: 3,
Ballot: Ballot{Replica: 2},
Deps: rn.q.deps(),
Command: Command{
Payload: []byte("x"),
ConflictKeys: [][]byte{[]byte("k")},
},
})
// Install the resident record in the engine so ProvideRecordLoad can find it.
foldTestRef(&rn.engine, stored)
// Build a load request with zero Ref that should be overridden by the result Ref.
rec := stored
rec.Ref = InstanceRef{} // zero Ref triggers override behavior
lane := 0 // use the default lane for this test
loaded, err := rn.engine.ProvideRecordLoad(lane, rec)
if err != nil {
t.Fatalf("ProvideRecordLoad unexpected error: %v", err)
}
if loaded.Ref != ref {
t.Fatalf("expected loaded.Ref to be %v, got %v", ref, loaded.Ref)
}
// Verify that the input record Ref was also updated to match the loaded Ref.
if rec.Ref != ref {
t.Fatalf("expected rec.Ref to be updated to %v, got %v", ref, rec.Ref)
}
}
func TestProvideRecordLoadMismatchedRefError(t *testing.T) {
rn, err := NewRawNode(Config{ID: 1, Voters: makeIDs(3)})
if err != nil {
t.Fatal(err)
}
// Create a resident record with a specific Ref.
ref := InstanceRef{Conf: 1, Replica: 2, Instance: 5}
stored := checkedRecord(InstanceRecord{
Ref: ref,
Status: StatusCommitted,
Seq: 3,
Ballot: Ballot{Replica: 2},
Deps: rn.q.deps(),
Command: Command{
Payload: []byte("x"),
ConflictKeys: [][]byte{[]byte("k")},
},
})
// Install the resident record in the engine so ProvideRecordLoad can find it.
foldTestRef(&rn.engine, stored)
// Build a load request whose Ref does not match the stored Ref.
rec := stored
rec.Ref = InstanceRef{Conf: 9, Replica: 9, Instance: 9}
lane := 0
_, err = rn.engine.ProvideRecordLoad(lane, rec)
if err == nil {
t.Fatalf("expected ProvideRecordLoad to fail for mismatched Ref, got nil error")
}
}- These tests assume
ProvideRecordLoadis a method with signaturefunc (e *engine) ProvideRecordLoad(lane int, rec InstanceRecord) (InstanceRecord, error). If the actual receiver type, lane type, parameter passing (by value vs pointer), or return values differ, adjust the call sites accordingly. - If
foldTestRefis not sufficient to make the record discoverable byProvideRecordLoad, you may need to mirror whatever setup otherProvideRecordLoadtests use (e.g., calling specific engine helpers to install the resident record). - If
InstanceRefcomparison requires a helper (e.g.,EqualInstanceRef) instead of!=/==, replace the direct comparisons with that helper to stay consistent with existing conventions. - If
record_load_test.godoes not yet import any additional packages (likereflect) and you choose to use them for comparisons, add the imports at the top of the file. - If
TestRecordLoadFoundReplaysDeferredcontains additional code after the snippet you provided, move the newly added tests so they are defined at file scope (i.e., after the closing brace of that test) to keep the file compiling.
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request replaces the O(all instances) conflict and attributes machinery with a per-lane conflict engine, introduces a two-tier executed-instance retirement mechanism with configurable per-lane retention, and implements an asynchronous record-load handshake for folded instances. The feedback is highly constructive and focuses on key performance and robustness improvements: optimizing the retireExecuted function to avoid O(N) map scans, validating returned record references in ProvideRecordLoad to prevent state corruption, avoiding redundant radix tree operations in conflictEngine.apply, simplifying capacity checks in deferRecordLoad, and safeguarding radixLevel against potential bit-shift overflows.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| type item struct { | ||
| ref InstanceRef | ||
| rec InstanceRecord | ||
| } | ||
| byLane := make(map[instanceLane][]item) | ||
| for ref, inst := range n.instances { | ||
| if inst == nil || inst.rec.Status != StatusExecuted || !n.executed.contains(ref) { | ||
| continue | ||
| } | ||
| lane := laneFor(ref) | ||
| byLane[lane] = append(byLane[lane], item{ref: ref, rec: inst.rec}) | ||
| } | ||
| for lane, items := range byLane { | ||
| executedThrough := n.executed.prefix(lane) | ||
| if executedThrough == 0 { | ||
| continue | ||
| } | ||
| var target InstanceNum | ||
| if executedThrough > retain { | ||
| target = executedThrough - retain | ||
| } | ||
| folded := n.engine.foldedThrough(lane) | ||
| if target <= folded { | ||
| continue | ||
| } | ||
| for _, it := range items { | ||
| if it.ref.Instance <= folded || it.ref.Instance > target { | ||
| continue | ||
| } | ||
| if n.instances[it.ref] == nil { | ||
| continue | ||
| } | ||
| rec := it.rec.Clone() | ||
| n.engine.foldRecord(rec) | ||
| delete(n.instances, it.ref) | ||
| n.foldedInstances++ | ||
| } |
There was a problem hiding this comment.
The current implementation of retireExecuted performs a full scan over the entire n.instances map on every Advance call. As the number of resident instances grows (up to MaxResidentInstances), this (folded, target].
for lane, index := range n.engine.laneIndex {
executedThrough := n.executed.prefix(lane)
if executedThrough == 0 {
continue
}
var target InstanceNum
if executedThrough > retain {
target = executedThrough - retain
}
folded := index.folded
if target <= folded {
continue
}
for instNum := folded + 1; instNum <= target; instNum++ {
ref := InstanceRef{Conf: lane.conf, Replica: lane.replica, Instance: instNum}
inst := n.instances[ref]
if inst == nil {
continue
}
if inst.rec.Status != StatusExecuted || !n.executed.contains(ref) {
continue
}
rec := inst.rec.Clone()
n.engine.foldRecord(rec)
delete(n.instances, ref)
n.foldedInstances++
}| if rec.Ref != res.Ref && !res.Ref.IsZero() { | ||
| // allow Ref on result to define | ||
| if rec.Ref.IsZero() { | ||
| rec.Ref = res.Ref | ||
| } | ||
| } |
There was a problem hiding this comment.
If the record returned by the embedding has a different Ref than requested (and is not zero), the current code silently proceeds, which can cause the requested res.Ref to remain uninstalled while another instance is modified. This will lead to infinite request loops or state corruption when replaying deferred messages. We should explicitly validate that the returned record's Ref matches the requested Ref.
if rec.Ref.IsZero() {
rec.Ref = res.Ref
} else if rec.Ref != res.Ref {
return fmt.Errorf("%w: record Ref %s does not match requested Ref %s", ErrInvalidRecord, rec.Ref, res.Ref)
}| func (e *conflictEngine) apply(prev *InstanceRecord, rec InstanceRecord) { | ||
| if prev != nil { | ||
| e.remove(prev.Ref, *prev) | ||
| } | ||
| lane := laneFor(rec.Ref) | ||
| index := e.ensureLane(lane) | ||
| _, existed := index.resident.slot(rec.Ref.Instance) | ||
| index.resident.set(rec.Ref.Instance, slotForRecord(rec)) | ||
| if !existed { | ||
| e.resident++ | ||
| } | ||
| if recordConflictEligible(rec) && !commandHasGlobalConflictScope(rec.Command.Kind) { | ||
| for _, key := range rec.Command.ConflictKeys { | ||
| e.ensureKeyLane(rec.Ref.Conf, key, lane).postings.insert(rec.Ref.Instance) | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
When updating an existing instance record (where prev != nil and prev.Ref == rec.Ref), the current implementation deletes the instance from the radix tree and postings tree, and then immediately re-inserts it. Since the conflict keys of an instance are immutable, this results in highly redundant radix tree deletions and insertions on every state transition. We can optimize this by updating the resident tree directly and only modifying the postings tree if the eligibility status actually changes.
func (e *conflictEngine) apply(prev *InstanceRecord, rec InstanceRecord) {
lane := laneFor(rec.Ref)
index := e.ensureLane(lane)
if prev != nil && prev.Ref != rec.Ref {
e.remove(prev.Ref, *prev)
prev = nil
}
_, existed := index.resident.slot(rec.Ref.Instance)
index.resident.set(rec.Ref.Instance, slotForRecord(rec))
if !existed {
e.resident++
}
prevEligible := prev != nil && recordConflictEligible(*prev) && !commandHasGlobalConflictScope(prev.Command.Kind)
currEligible := recordConflictEligible(rec) && !commandHasGlobalConflictScope(rec.Command.Kind)
if prevEligible && !currEligible {
for _, key := range prev.Command.ConflictKeys {
e.removeKeyPosting(prev.Ref.Conf, key, lane, prev.Ref.Instance)
}
} else if !prevEligible && currEligible {
for _, key := range rec.Command.ConflictKeys {
e.ensureKeyLane(rec.Ref.Conf, key, lane).postings.insert(rec.Ref.Instance)
}
}
}| if n.pendingRecordLoads == nil { | ||
| n.pendingRecordLoads = make(map[InstanceRef]*recordLoadWait) | ||
| } | ||
| wait := n.pendingRecordLoads[ref] | ||
| if wait == nil { | ||
| if n.pendingRecordLoadMessages >= n.maxDeferredRecordLoads { | ||
| return fmt.Errorf("%w: %w", ErrMessageRejected, ErrDeferredRecordLoadFull) | ||
| } | ||
| wait = &recordLoadWait{} | ||
| n.pendingRecordLoads[ref] = wait | ||
| // enqueue request into next Ready batch | ||
| target := n.readyTarget() | ||
| target.RecordLoads = append(target.RecordLoads, ref) | ||
| // keep sorted unique later at Ready() freeze; also sort now for stability | ||
| sortRefs(target.RecordLoads) | ||
| // dedup adjacent | ||
| out := target.RecordLoads[:0] | ||
| var last InstanceRef | ||
| for i, r := range target.RecordLoads { | ||
| if i == 0 || r != last { | ||
| out = append(out, r) | ||
| last = r | ||
| } | ||
| } | ||
| target.RecordLoads = out | ||
| wait.requested = true | ||
| } | ||
| if n.pendingRecordLoadMessages >= n.maxDeferredRecordLoads { | ||
| return fmt.Errorf("%w: %w", ErrMessageRejected, ErrDeferredRecordLoadFull) | ||
| } |
There was a problem hiding this comment.
The capacity check n.pendingRecordLoadMessages >= n.maxDeferredRecordLoads is duplicated inside the wait == nil block and at the end of the function. We can simplify the control flow and avoid redundant map lookups/insertions by checking the capacity limit once at the very beginning of deferRecordLoad.
if n.pendingRecordLoadMessages >= n.maxDeferredRecordLoads {
return fmt.Errorf("%w: %w", ErrMessageRejected, ErrDeferredRecordLoadFull)
}
if n.pendingRecordLoads == nil {
n.pendingRecordLoads = make(map[InstanceRef]*recordLoadWait)
}
wait := n.pendingRecordLoads[ref]
if wait == nil {
wait = &recordLoadWait{}
n.pendingRecordLoads[ref] = wait
// enqueue request into next Ready batch
target := n.readyTarget()
target.RecordLoads = append(target.RecordLoads, ref)
// keep sorted unique later at Ready() freeze; also sort now for stability
sortRefs(target.RecordLoads)
// dedup adjacent
out := target.RecordLoads[:0]
var last InstanceRef
for i, r := range target.RecordLoads {
if i == 0 || r != last {
out = append(out, r)
last = r
}
}
target.RecordLoads = out
wait.requested = true
}| func radixLevel(value InstanceNum) uint8 { | ||
| for level := uint8(0); level < maxRadixLevel; level++ { | ||
| if uint64(value) < uint64(1)<<(radixBits*(level+1)) { | ||
| return level | ||
| } | ||
| } | ||
| return maxRadixLevel | ||
| } |
There was a problem hiding this comment.
The radixLevel function shifts uint64(1) by radixBits * (level + 1). If maxRadixLevel or radixBits is increased in the future such that the shift count equals or exceeds 64, this will cause undefined behavior or incorrect level computation. We should add a safety check to break early if the shift count is >= 64 to make the function robust and future-proof.
func radixLevel(value InstanceNum) uint8 {
for level := uint8(0); level < maxRadixLevel; level++ {
shift := radixBits * (level + 1)
if shift >= 64 {
break
}
if uint64(value) < uint64(1)<<shift {
return level
}
}
return maxRadixLevel
}![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: d7241afd7d
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| MustSync bool | ||
| // RecordLoads requests durable InstanceRecords for folded refs referenced by | ||
| // inbound messages. Sorted and deduplicated; stable until Advance. | ||
| RecordLoads []InstanceRef |
There was a problem hiding this comment.
Adding RecordLoads here also requires Ready.Empty to treat them as work. As written, a deferred folded-record lookup only appends RecordLoads and leaves MustSync false, so HasReady stays false, ReadyInto returns an empty batch, and Advance rejects a RecordLoads-only acknowledgement as empty. This makes the asynchronous load handshake unreachable whenever a folded ref is first requested.
Useful? React with 👍 / 👎.
| n.pendingRecordLoads[ref] = wait | ||
| // enqueue request into next Ready batch | ||
| target := n.readyTarget() | ||
| target.RecordLoads = append(target.RecordLoads, ref) |
There was a problem hiding this comment.
When awaitAdvance is true, readyTarget() points at nextReady, so this append can queue a load request behind the currently frozen batch. mergeNextReady never moves nextReady.RecordLoads into pendingReady, so after the frozen Ready is advanced the request disappears while pendingRecordLoads still holds the deferred messages; no subsequent Ready asks the embedding to load that record, leaving those messages stuck until the deferred-load capacity is exhausted.
Useful? React with 👍 / 👎.
| if n.needsRecordLoad(m.Ref) { | ||
| return n.deferRecordLoad(m.Ref, m) |
There was a problem hiding this comment.
This folded-record load guard is only added for Prepare/Evidence. If a delayed MsgPreAccept or MsgAccept arrives for an already executed ref at or below foldedThrough, handlePreAccept/handleAccept still see n.instances[m.Ref] == nil and install a fresh PreAccepted/Accepted record over the folded executed summary instead of loading the executed record and replying with the commit, which can regress durable status for a completed instance.
Useful? React with 👍 / 👎.
| if rec.Ref != res.Ref && !res.Ref.IsZero() { | ||
| // allow Ref on result to define | ||
| if rec.Ref.IsZero() { | ||
| rec.Ref = res.Ref | ||
| } |
There was a problem hiding this comment.
When Found is true and res.Record.Ref is set to a different non-zero ref, this branch does not reject it; the checksum is then validated for the unrelated record, that record is installed, and the wait for res.Ref is cleared. A stale or mis-keyed storage response can therefore corrupt an unrelated instance while replaying the original deferred messages without their requested folded record; return ErrInvalidRecord unless the loaded record ref is zero or exactly matches res.Ref.
Useful? React with 👍 / 👎.
| len(rd.Messages) > len(n.frozenReady.Messages) || | ||
| len(rd.BootstrapMessages) > len(n.frozenReady.BootstrapMessages) || | ||
| len(rd.Committed) > len(n.frozenReady.Committed) || | ||
| len(rd.RecordLoads) > len(n.frozenReady.RecordLoads) || |
There was a problem hiding this comment.
This only bounds the number of acknowledged RecordLoads; unlike the other Ready slices validated below, it never compares the actual refs against n.frozenReady.RecordLoads. If an embedding reuses or mutates the Ready and advances a same-length but different load list, Advance consumes the real frozen request and the original folded messages remain waiting for a record the embedding was no longer asked to fetch.
Useful? React with 👍 / 👎.
| n.engine.foldRecord(rec) | ||
| delete(n.instances, ref) |
There was a problem hiding this comment.
When ProvideRecordLoad installs an executed folded record, installInstance adds it back to executed.exact even though executed.through already covers that ref. This refold path deletes the resident again but never removes the redundant exact entry, so repeated loads of old folded refs grow the exact map and undermine the retention bound.
Useful? React with 👍 / 👎.
|
|
||
| n.enqueueExecutedRecords(rd.Committed[:ackedCommitted]) | ||
| n.applyBootstrapDurability(rd.BootstrapRecords) | ||
| n.retireExecuted() |
There was a problem hiding this comment.
For user commands, enqueueExecutedRecords only places the StatusExecuted record into the next Ready batch; this new call folds and deletes the resident before that marker has been persisted and advanced. If a late Prepare/Evidence arrives in that window, the node asks storage for the folded ref and can load the still-committed durable record, then replaying the deferred message can enqueue that committed record after the pending executed marker and leave the instance regressed to committed. Retire only records whose executed marker has already been acknowledged durable.
Useful? React with 👍 / 👎.
fc7391b to
fcf1ffe
Compare
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
1 participant
Solution / proposal
retireExecuted folds contiguous executed prefixes via engine summaries; executedTracker.contains includes through[lane]; Propose MaxResidentInstances backpressure.
Closes #
Summary by Sourcery
Replace the legacy conflict indexes with a structured per-lane conflict engine, add executed-instance retirement backed by an asynchronous record-load handshake and resident backpressure, harden decoding and tooling around safety and validation, and wire linting plus expanded tests and docs into the build and CI pipeline.
New Features:
Bug Fixes:
Enhancements:
Build:
CI:
Documentation:
Tests:
Chores:
Closes #12