Give each asking at unchanged heads its own router id, overwrite the ask instead of taking it away, and announce the answered asking (review round 1, G2; issue 1370)

This commit is contained in:
2026-10-20 01:26:28 +00:00
parent 67fe9a4dbe
commit 83adc46fe0
6 changed files with 153 additions and 15 deletions
@@ -350,12 +350,22 @@ func (h *Holder) approval(d *Delivery, now time.Time) (string, func()) {
return "the operator cannot be asked: " + err.Error(), nil
}
r.Asks++
ask := composeMergeAsk(sdk.AskID(id), d, group, members, heads, now)
// Each asking at these heads has its own router id (sdk.AskingID): the router keeps an ask by its id and drops a second ask
// under it as a redelivery, so an ask written again at unchanged heads (a lapse, a status or the review record changing) is
// the next asking, with a new id (review round 1, G2; issue 1370).
asking := 1
for _, m := range members {
if m.Ready != nil && m.Ready.AskedKey == id {
asking = max(asking, m.Ready.Asking+1)
}
}
ask := composeMergeAsk(sdk.AskingID(id, asking), d, group, members, heads, now)
taken := []*Delivery{}
var reads []memberRead
for _, m := range members {
h.runOf(m, now)
m.Ready.AskID, m.Ready.Ask, m.Ready.Heads, m.Ready.AskedAt, m.Ready.Asks = id, &ask, heads, now, max(m.Ready.Asks, r.Asks)
m.Ready.AskedKey, m.Ready.Asking = id, asking
h.keep(m)
taken = append(taken, m)
reads = append(reads, readOf(m, m.Commit))
@@ -393,7 +403,23 @@ func (h *Holder) approval(d *Delivery, now time.Time) (string, func()) {
}
}
}
// An earlier asking of this key may still stand in the bucket: not written again, it is taken away.
stale := func() {
h.mu.Lock()
standing := h.standing(id)
h.mu.Unlock()
if standing {
if err := h.Operator.Cancel(id); err != nil {
h.Logf("[mesh-delivery] the operator's ask %s could not be withdrawn (%v)", id, err)
return
}
h.mu.Lock()
h.markWritten(id, false)
h.mu.Unlock()
}
}
if readErr != nil || holdSays != "" {
stale()
h.mu.Lock()
defer h.mu.Unlock()
undo()
@@ -416,7 +442,11 @@ func (h *Holder) approval(d *Delivery, now time.Time) (string, func()) {
h.keepGroup(g)
h.mu.Unlock()
}
if err := h.Operator.Ask(id, sdk.Ask{Group: name, Heads: covered, Ask: ask}); err != nil {
if err := h.Operator.Ask(id, sdk.Ask{Group: name, Heads: covered, Asking: asking, Ask: ask}); err == nil {
h.mu.Lock()
h.markWritten(id, true)
h.mu.Unlock()
} else {
h.mu.Lock()
defer h.mu.Unlock()
undo()
@@ -460,7 +490,11 @@ func (h *Holder) dropAsk(id, why string) func() {
}
if err := h.Operator.Cancel(id); err != nil {
h.Logf("[mesh-delivery] the operator's ask %s could not be cancelled (%v)", id, err)
return
}
h.mu.Lock()
h.markWritten(id, false)
h.mu.Unlock()
}
}
@@ -474,15 +508,37 @@ func (h *Holder) withdrawAsk(id, why string) func() {
h.keep(d)
}
}
// The key is not taken away: the next look writes it again with the next asking, if the group may be asked, and takes it away
// only if it may not (readyAsks, withdrawnKeys), so the key is never absent between the two.
h.withdrawn[id] = true
h.Logf("[mesh-delivery] the operator's ask %s is withdrawn to be asked again: %s", id, why)
return func() {
if h.Operator == nil {
return
return func() {}
}
// withdrawnKeys are, after a look, the closures that take away the asks withdrawn and not written again by it. Under the lock.
func (h *Holder) withdrawnKeys() []func() {
var out []func()
for key := range h.withdrawn {
delete(h.withdrawn, key)
again := false
for _, d := range h.deliveries {
again = again || (d.Ready != nil && d.Ready.AskID == key)
}
if err := h.Operator.Cancel(id); err != nil {
h.Logf("[mesh-delivery] the operator's ask %s could not be withdrawn (%v)", id, err)
if again || h.Operator == nil {
continue
}
key := key
out = append(out, func() {
if err := h.Operator.Cancel(key); err != nil {
h.Logf("[mesh-delivery] the operator's ask %s could not be withdrawn (%v)", key, err)
return
}
h.mu.Lock()
h.markWritten(key, false)
h.mu.Unlock()
})
}
return out
}
// pendingAsks are the asks waiting for an answer, by id; under the lock.
@@ -563,6 +619,13 @@ func (h *Holder) answered(key string, ans sdk.Answer) []func() {
refuse(fmt.Sprintf("its key %s is not the key %s of the heads asked", key, want))
return nil
}
// The answer is to the asking asked now: the teller names the router id it was given in the warrant's id (`<ask id>@<time>`),
// and an answer to an earlier asking is never taken for a later one (issue 1370).
if now, at, ok := strings.Cut(ans.Warrant, "@"); ok && at != "" && now != sdk.AskingID(key, first.Asking) {
h.Logf("[mesh-delivery] the answer to the ask %s was given to an earlier asking (%s), not the one asked now (%s): not taken",
key, now, sdk.AskingID(key, first.Asking))
return nil
}
if ans.Option != OptionApprove && ans.Option != OptionDecline {
refuse("it chose " + ans.Option + ", which the ask did not offer")
return nil
@@ -618,3 +681,23 @@ func (kvOperator) Where(key string) (*sdk.Answer, error) {
}
return &a, nil
}
// standing is whether an ask is written under a key in the gate's asks, as this owner last knew. Under the lock.
func (h *Holder) standing(key string) bool {
for _, d := range h.deliveries {
if d.Ready != nil && d.Ready.AskedKey == key && d.Ready.Written {
return true
}
}
return false
}
// markWritten keeps whether an ask stands under a key, on every delivery that asked it. Under the lock.
func (h *Holder) markWritten(key string, written bool) {
for _, d := range h.deliveries {
if d.Ready != nil && d.Ready.AskedKey == key && d.Ready.Written != written {
d.Ready.Written = written
h.keep(d)
}
}
}
@@ -170,8 +170,13 @@ func TestAnUnansweredAskIsWrittenAgainOnceWhenItLapses(t *testing.T) {
first := w.op.open()[0]
w.later(AskLasts + time.Minute)
w.settleAll()
if w.op.count() != 2 || len(w.op.open()) != 1 || w.op.open()[0] != first || len(w.op.cancelled) != 1 || w.op.cancelled[0] != first {
t.Fatalf("a lapsed ask was not written again once: %d asks, open %v, taken away %v", w.op.count(), w.op.open(), w.op.cancelled)
if w.op.count() != 2 || len(w.op.open()) != 1 || w.op.open()[0] != first || len(w.op.cancelled) != 0 {
t.Fatalf("a lapsed ask was not written again once, over the same key: %d asks, open %v, taken away %v", w.op.count(), w.op.open(), w.op.cancelled)
}
// Review round 1, G2: the router keeps an ask by its id and drops a second under it, so the second asking has an id of its own.
again := w.op.byKey[first]
if again.Asking != 2 || again.Ask.ID != sdk.AskingID(first, 2) || again.Ask.ID == sdk.AskID(first) {
t.Fatalf("the second asking is %d under the id %s; wanted asking 2 under %s", again.Asking, again.Ask.ID, sdk.AskingID(first, 2))
}
for range 3 {
w.later(2 * time.Minute)
@@ -376,3 +381,28 @@ func TestIssue1025AnOwedMergedIsSaidAfterARestartAndOnce(t *testing.T) {
t.Fatalf("said %d times after the restart; wanted once", n)
}
}
// Review round 1, G2: an answer given to the first asking is not taken for the second, and the second's Approve is announced
// under the second asking's id.
func TestAnAnswerToAnEarlierAskingIsNotTakenAndTheMergeNamesTheAnsweredAsking(t *testing.T) {
w := markedGreen(t)
key := w.op.open()[0]
w.later(AskLasts + time.Minute)
w.settleAll() // asked again: the second asking
w.op.mu.Lock()
w.op.answers[key] = &sdk.Answer{Option: OptionApprove, Warrant: sdk.AskingID(key, 1) + "@2026-10-06T12:00:00Z", Words: "the operator, via telegram, chose Approve"}
w.op.mu.Unlock()
w.settleAll()
if len(w.forge.merges) != 0 || len(w.merges.puts) != 0 {
t.Fatalf("an answer to the first asking merged on the second: %v", w.forge.merges)
}
w.op.answer(key, OptionApprove) // the answer to the asking asked now
w.settleAll()
if len(w.forge.merges) != 1 {
t.Fatalf("the answer to the second asking did not merge: %v", w.forge.merges)
}
say := mergedSaid(w)
if len(say) != 1 || say[0]["ask_id"] != sdk.AskingID(key, 2) {
t.Fatalf("the merge is announced %v; wanted the ask id of asking 2, %s", say, sdk.AskingID(key, 2))
}
}
@@ -536,8 +536,13 @@ func TestAPassingVerdictAsksTheApproveAndItsApproveMergesWithinSecondsOverARealB
a, ok := askNow(b, key) // the key may be absent for an instant while the ask is taken away and written again
return ok && strings.Contains(a.Ask.Explanation, "reviewed by lead on shanks")
})
if h, err := b.kvs["asks"].History(ctx, key); err != nil || len(h) != 3 || h[1].Operation() != jetstream.KeyValueDelete {
t.Fatalf("the ask's history is %d revisions (%v); wanted written, taken away, written", len(h), err)
// The ask is overwritten, never taken away: the key is there at every revision, and the second asking has an id of its own
// (review round 1, G2), which the router would take as new.
if h, err := b.kvs["asks"].History(ctx, key); err != nil || len(h) != 2 || h[1].Operation() != jetstream.KeyValuePut {
t.Fatalf("the ask's history is %d revisions (%v); wanted written and written again", len(h), err)
}
if again := askAt(t, b, key); again.Asking != 2 || again.Ask.ID != sdk.AskingID(key, 2) || again.Ask.ID == ask.Ask.ID {
t.Fatalf("the second asking is %d under %s, the first was %s", again.Asking, again.Ask.ID, ask.Ask.ID)
}
if len(rt.asksOf("gitea.gitea_merge_pull_request")) != 0 {
t.Fatalf("merged before the Approve")
@@ -613,6 +618,15 @@ func askNow(b *testBus, key string) (sdk.Ask, bool) {
return a, err == nil
}
func askNowOrFatal(t *testing.T, b *testBus, key string) sdk.Ask {
t.Helper()
a, ok := askNow(b, key)
if !ok {
t.Fatalf("no ask under %s to answer", key)
}
return a
}
func askAt(t *testing.T, b *testBus, key string) sdk.Ask {
t.Helper()
e, err := b.kvs["asks"].Get(context.Background(), key)
@@ -638,7 +652,8 @@ func mergeAt(b *testBus, group string) sdk.Merge {
// answer is the teller's answer to an ask, written under the ask's own key in its bucket answers.
func (b *testBus) answer(t *testing.T, key, option string) time.Time {
t.Helper()
raw, err := sdk.EncodeAnswer(sdk.Answer{Option: option, Warrant: "w-" + key, Words: "the operator, via telegram, chose " + option})
// The warrant's id is the router id of the asking answered and the time (the teller's warrantID).
raw, err := sdk.EncodeAnswer(sdk.Answer{Option: option, Warrant: askNowOrFatal(t, b, key).Ask.ID + "@2026-10-20T03:00:00Z", Words: "the operator, via telegram, chose " + option})
if err != nil {
t.Fatal(err)
}
@@ -45,6 +45,9 @@ type Holder struct {
// cancels are asks to take back, gathered under the lock and run with the other asks of the look.
cancels []func()
// withdrawn are the ask keys withdrawn since the last look (approval.go withdrawAsk).
withdrawn map[string]bool
seatHeld atomic.Bool
seatMu sync.Mutex
seatWhy string
@@ -154,6 +157,7 @@ func (h *Holder) init() {
h.deliveries = map[string]*Delivery{}
h.groups = map[string]*Group{}
h.dirty = map[string]bool{}
h.withdrawn = map[string]bool{}
h.written = map[string][]byte{}
h.dirtyGroup = map[string]bool{}
h.unmatched = map[string]unmatched{}
@@ -44,8 +44,8 @@ func TestAReviewedByLineWrittenAfterTheAskWithdrawsTheUnansweredAskAndAsksAgain(
if len(open) != 1 || open[0] != first || w.op.count() != 2 {
t.Fatalf("after the edit the open asks are %v (first %s), %d asked; wanted the first withdrawn and written again once", open, first, w.op.count())
}
if len(w.op.cancelled) != 1 || w.op.cancelled[0] != first {
t.Fatalf("the first ask was not withdrawn: %v", w.op.cancelled)
if len(w.op.cancelled) != 0 || w.op.byKey[first].Asking != 2 {
t.Fatalf("the first ask was taken away (%v) or not written again as asking 2 (%d)", w.op.cancelled, w.op.byKey[first].Asking)
}
if e := w.op.ask(open[0]).Explanation; !strings.Contains(e, "reviewed by reviewer on shanks at "+shortOf(readyHead, 12)) ||
strings.Contains(e, "not reviewed") {
@@ -64,6 +64,11 @@ type ReadyRun struct {
merging bool
// refusedNoted is whether the refusal was written into the group's merge key. Not kept.
refusedNoted bool
// AskedKey and Asking are the ask key last asked for and how many times it was asked at those heads; they stay when the ask
// is withdrawn, so the next asking is the next number (sdk.AskingID). Written is whether that ask stands in the gate's asks.
AskedKey string `json:"asked_key,omitempty"`
Asking int `json:"asking,omitempty"`
Written bool `json:"written,omitempty"`
// MergeCommit is the commit the forge bridge reported the merge as, and SourceMoved whether source.main holds it
// (merges.go).
MergeCommit string `json:"merge_commit,omitempty"`
@@ -225,6 +230,7 @@ func (h *Holder) readyAsks(now time.Time) []func() {
}
}
asks = append(asks, h.settleMerges()...)
asks = append(asks, h.withdrawnKeys()...)
asks = append(h.cancels, asks...)
h.cancels = nil
return asks
@@ -411,7 +417,7 @@ func mergedEvent(d *Delivery, head string) map[string]any {
ev["ask_id"], ev["option"] = r.AskID, OptionApprove
if _, _, _, err := sdk.ParseAskKey(r.AskID); err == nil {
// The ask of the gate: the router knows it by the id its key gives, and the teller asked it (ADR 0437 §6).
ev["ask_id"], ev["asker"] = sdk.AskID(r.AskID), sdk.TellerModule
ev["ask_id"], ev["asker"] = sdk.AskingID(r.AskID, r.Asking), sdk.TellerModule
} // an Approve carried over from today's module keeps its router id, and the asker the controller takes by default
}
return ev