From 83adc46fe08d1ffb319d9803c827b784a9ad8636 Mon Sep 17 00:00:00 2001 From: mesh-tools Date: Tue, 20 Oct 2026 01:26:28 +0000 Subject: [PATCH] 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) --- .../cmd/mesh-delivery-gate/approval.go | 97 +++++++++++++++++-- .../cmd/mesh-delivery-gate/approval_test.go | 34 ++++++- .../cmd/mesh-delivery-gate/bus_test.go | 21 +++- .../cmd/mesh-delivery-gate/holder.go | 4 + .../cmd/mesh-delivery-gate/issue1079_test.go | 4 +- .../cmd/mesh-delivery-gate/ready.go | 8 +- 6 files changed, 153 insertions(+), 15 deletions(-) diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/approval.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/approval.go index 3fbb0070..71c3fc5b 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/approval.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/approval.go @@ -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 (`@