Keep a question's slot until it expires while the router has not recorded it (issue 589, review 107)
mesh/merge-gate head c4eebce598cb merged into trunk 9c9a76d220f8 — pass: builds claude-code → ace, g14, novox, shanks; no bus step; every machine composes …
mesh/repo-check head c4eebce598cb merged into trunk 9c9a76d220f8 — pass: its merge-check.sh passed, each part in its toolchain: merge-check.sh (go), merge-…
mesh/delivery delivered
mesh/merge-gate head c4eebce598cb merged into trunk 9c9a76d220f8 — pass: builds claude-code → ace, g14, novox, shanks; no bus step; every machine composes …
mesh/repo-check head c4eebce598cb merged into trunk 9c9a76d220f8 — pass: its merge-check.sh passed, each part in its toolchain: merge-check.sh (go), merge-…
mesh/delivery delivered
Freeing the slot two minutes after asking could let a backlogged router still show the question after the session read it as ended, and let a third question in. The seat's queue offers work again after an error and the router drops a cancel of an ask it has not recorded, so a cancel cannot take it back; once expired, the router refuses it if taken late. The slot test now uses a stale revision, so a plain write in place of the compare-and-set fails it.
This commit is contained in:
@@ -314,6 +314,8 @@ type memQuestions struct {
|
||||
n uint64
|
||||
// failGet fails every read of a key with this prefix.
|
||||
failGet string
|
||||
// afterGet runs after a read, outside the lock: another node writing between a read and a write.
|
||||
afterGet func(key string)
|
||||
}
|
||||
|
||||
func newMemQuestions() *memQuestions {
|
||||
@@ -322,12 +324,17 @@ func newMemQuestions() *memQuestions {
|
||||
|
||||
func (s *memQuestions) Get(key string) (json.RawMessage, uint64, bool, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if s.failGet != "" && strings.HasPrefix(key, s.failGet) {
|
||||
s.mu.Unlock()
|
||||
return nil, 0, false, errors.New("the bus did not answer")
|
||||
}
|
||||
raw, ok := s.m[key]
|
||||
return raw, s.rev[key], ok, nil
|
||||
rev, after := s.rev[key], s.afterGet
|
||||
s.mu.Unlock()
|
||||
if after != nil {
|
||||
after(key)
|
||||
}
|
||||
return raw, rev, ok, nil
|
||||
}
|
||||
func (s *memQuestions) put(key string, value any) error {
|
||||
raw, err := json.Marshal(value)
|
||||
@@ -431,22 +438,60 @@ func TestTheCapFailsClosed(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// A question whose refusal was not heard here, and of which the router holds no record, frees its slot once the
|
||||
// router has had time to record it, and reads as ended — not open until it expires.
|
||||
func TestAQuestionTheRouterNeverHeldFreesItsSlot(t *testing.T) {
|
||||
// A question whose refusal was not heard here, and of which the router holds no record, keeps its slot and reads as
|
||||
// waiting until it expires — the router may still take it from its queue — and only then frees its slot and ends.
|
||||
func TestAQuestionTheRouterHasNotRecordedHoldsItsSlotUntilItExpires(t *testing.T) {
|
||||
p, _ := node(t, "g14")
|
||||
store, r := newMemQuestions(), newRouter()
|
||||
first := askMergeTonight(t, p, store, r, t0)
|
||||
askMergeTonight(t, p, store, r, t0)
|
||||
delete(r.records, first) // refused, and the refusal never heard here
|
||||
if _, err := AskQuestion(p, mergeTonight(), store, r, t0.Add(time.Minute)); err == nil {
|
||||
t.Fatal("a slot was freed before the router had time to record the question")
|
||||
delete(r.records, first) // the router down or backlogged, or its refusal never heard here
|
||||
later := t0.Add(QuestionDefault - time.Minute)
|
||||
if _, err := AskQuestion(p, mergeTonight(), store, r, later); err == nil {
|
||||
t.Fatal("the slot of a question the router may still show was freed before it expired")
|
||||
}
|
||||
if out, _ := ReadAnswer(store, r, first, t0.Add(RouterHoldsWithin)); out["state"] != AnswerEnded {
|
||||
t.Errorf("read as %v", out)
|
||||
out, _ := ReadAnswer(store, r, first, later)
|
||||
if out["state"] != AnswerWaiting || !strings.Contains(out["words"].(string), "not yet recorded") {
|
||||
t.Errorf("before expiry read as %v", out)
|
||||
}
|
||||
if _, err := AskQuestion(p, mergeTonight(), store, r, t0.Add(RouterHoldsWithin)); err != nil {
|
||||
t.Errorf("the slot of a question the router never held was not freed: %v", err)
|
||||
expired := t0.Add(QuestionDefault)
|
||||
if out, _ := ReadAnswer(store, r, first, expired); out["state"] != AnswerEnded ||
|
||||
out["now"] != "do what you said: I leave the pull request open" {
|
||||
t.Errorf("at expiry read as %v", out)
|
||||
}
|
||||
if _, err := AskQuestion(p, mergeTonight(), store, r, expired); err != nil {
|
||||
t.Errorf("the slot of an expired question was not freed: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// A slot is taken only at the revision read: when another node takes both slots between this node's read and its
|
||||
// write, this node's writes are refused and the question is refused — a plain write would take a slot it lost.
|
||||
func TestASlotIsTakenOnlyAtTheRevisionRead(t *testing.T) {
|
||||
p, _ := node(t, "g14")
|
||||
store, r := newMemQuestions(), newRouter()
|
||||
// Two questions that ended: their slots are free to take again.
|
||||
for _, id := range []string{askMergeTonight(t, p, store, r, t0), askMergeTonight(t, p, store, r, t0)} {
|
||||
r.close(asks.Warrant{Ask: id, Outcome: asks.OutcomeExpired, At: t0.Add(time.Minute)})
|
||||
}
|
||||
shanks, _ := node(t, "shanks")
|
||||
other := newRouter()
|
||||
store.afterGet = func(key string) {
|
||||
if !strings.HasPrefix(key, "slot-") {
|
||||
return
|
||||
}
|
||||
// Another node asks between this node's read of the slot and its write: its question waits.
|
||||
q := Question{ID: newQuestionID(), Headline: "Another", Context: "c", Why: "w", OnExpiry: "e", By: "other",
|
||||
From: shanks.Node, At: t0, Expires: t0.Add(time.Hour),
|
||||
Options: []QuestionOption{{ID: "o1", Label: "A", Means: "a", Recommended: true}, {ID: "o2", Label: "B", Means: "b"}}}
|
||||
_ = store.Put(q.ID, q)
|
||||
_ = other.Publish(asks.AskSubject(Module), q.Ask(), q.ID)
|
||||
r.records[q.ID] = other.records[q.ID]
|
||||
_ = store.Put(key, slot{Question: q.ID, At: t0})
|
||||
}
|
||||
_, err := AskQuestion(p, mergeTonight(), store, r, t0.Add(2*time.Minute))
|
||||
store.afterGet = nil
|
||||
if err == nil || !strings.Contains(err.Error(), "questions already wait") {
|
||||
t.Fatalf("a slot lost to another node was taken: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user