Kaien
af2560222f
Fence stale ACK results by generation, make duplicate MarkProcessed harmless after confirmed ACK, and cover forged delivery, mutation, in-flight ACK, initial subscribe, and replacement edge cases. Co-authored-by: Cursor <cursoragent@cursor.com>
76 lines
1.5 KiB
Go
76 lines
1.5 KiB
Go
package chatclient
|
|
|
|
type deliveryToken struct {
|
|
_ byte
|
|
}
|
|
|
|
type ackEntry struct {
|
|
seq int64
|
|
token *deliveryToken
|
|
done bool
|
|
}
|
|
|
|
type ackTracker struct {
|
|
entries []ackEntry
|
|
index map[*deliveryToken]int
|
|
retired map[*deliveryToken]struct{}
|
|
}
|
|
|
|
func newACKTracker() ackTracker {
|
|
return ackTracker{
|
|
index: make(map[*deliveryToken]int),
|
|
retired: make(map[*deliveryToken]struct{}),
|
|
}
|
|
}
|
|
|
|
func (t *ackTracker) add(seq int64, token *deliveryToken) {
|
|
t.index[token] = len(t.entries)
|
|
t.entries = append(t.entries, ackEntry{seq: seq, token: token})
|
|
}
|
|
|
|
func (t *ackTracker) mark(token *deliveryToken) (int64, bool, bool) {
|
|
i, ok := t.index[token]
|
|
if !ok {
|
|
if _, retired := t.retired[token]; retired {
|
|
return t.safeBoundary(), true, true
|
|
}
|
|
return 0, false, false
|
|
}
|
|
if t.entries[i].done {
|
|
return t.safeBoundary(), true, true
|
|
}
|
|
t.entries[i].done = true
|
|
return t.safeBoundary(), true, true
|
|
}
|
|
|
|
func (t *ackTracker) safeBoundary() int64 {
|
|
var boundary int64
|
|
for i := range t.entries {
|
|
if !t.entries[i].done {
|
|
break
|
|
}
|
|
boundary = t.entries[i].seq
|
|
}
|
|
return boundary
|
|
}
|
|
|
|
func (t *ackTracker) removeConfirmed(lastAck int64) {
|
|
n := 0
|
|
for n < len(t.entries) {
|
|
if !t.entries[n].done || t.entries[n].seq > lastAck {
|
|
break
|
|
}
|
|
t.retired[t.entries[n].token] = struct{}{}
|
|
delete(t.index, t.entries[n].token)
|
|
n++
|
|
}
|
|
if n == 0 {
|
|
return
|
|
}
|
|
copy(t.entries, t.entries[n:])
|
|
t.entries = t.entries[:len(t.entries)-n]
|
|
for i := range t.entries {
|
|
t.index[t.entries[i].token] = i
|
|
}
|
|
}
|