Retire the controller's merges consumer and read merges from the forge bridge's stream, so a held merge can never again queue every merge or look like a deaf controller (issue 774)
The controller-merges consumer (no max-deliver, one at a time) held the de-spiegel merge e25e1beb from 10-29 09:39: keepForgeMerge put an announcement with its Paths into delivery-merges (16 KiB values), got 10054, retried "not now" forever, and 167 merges queued behind it. Every five-minute hold outran the deaf bound, so core.controller.deaf flapped. Merges are mesh-delivery's: the gate merges on the Approve and the forge bridge says each merge on its stream FORGE (pull request, head, merge commit), kept a week. The controller's copy was a leftover, read only at a moved module's registration (ADR 0420 §7, ADR 0432). That read now finds the merge on FORGE; the consumer, its follow, its handler and its hold delays are removed, and every assertion removes the retired consumer from the bus. Kind fix for the watchdog: the event loop records when a message it held back is due again, and deafness counts from then, so no hold longer than the bound (on any consumer) reads as deafness. Quick fix: strip Paths from the kept announcement and add max-deliver with a dead letter. Thorough fix: remove the duplicate duty and make the deaf watchdog hold-aware. Chose thorough: the controller is being deprecated and the merges already have an owner; the hold-vs-deaf confusion is a kind, not one consumer. Research 072 piece: 03-domains (Control shrinks to placement and composition; the delivery parts own what a merge is) and milestone 3 (health signals stop sharing the controller's loop assumptions).
This commit is contained in:
+34
-22
@@ -339,13 +339,36 @@ func Overlaps() []string {
|
||||
// ack subject is derived from (nats.go: `$JS.ACK.<stream>.controller.>`).
|
||||
const ControllerName = "controller"
|
||||
|
||||
// MergesConsumerName is the controller's durable consumer of the forge's merged event on EVENTS, apart from the
|
||||
// events consumer (novox/hq ADR 0407 decision 3, issue 973). A merge is never dropped, so this consumer has no
|
||||
// max-deliver, which would send a merge held five times to DEAD_LETTERS to wait for a person, and no healer
|
||||
// resets it, which would drop the merges it holds. A consumer of its own because both of those are properties of
|
||||
// a consumer, not of a subject.
|
||||
// MergesConsumerName was the controller's durable consumer of the forge's merged event on EVENTS (novox/hq ADR 0407
|
||||
// decision 3, issue 973), with no max-deliver and one message at a time. **Retired** (sweep 2026-10-30, issue 774):
|
||||
// the forge bridge's stream FORGE is the record of the merges, and the controller reads it where it needs a merge. On
|
||||
// 2026-10-29 one announcement too large for the controller's bucket of merges was held on it for over a day, with
|
||||
// 163 merges queued behind it, and the deaf watchdog read the queue as a deaf controller. RetiredConsumers removes it.
|
||||
const MergesConsumerName = "controller-merges"
|
||||
|
||||
// RetiredConsumers are durable consumers the controller once held and holds no more: each assertion removes them,
|
||||
// so a retired consumer cannot keep a queue nobody takes from (research 072, 03-domains: the controller shrinks).
|
||||
var RetiredConsumers = []Consumer{{Name: MergesConsumerName, Stream: EventsStream}}
|
||||
|
||||
// Remover is the part of a JetStream connection that removes a retired consumer.
|
||||
type Remover interface {
|
||||
RemoveConsumer(stream, name string) error
|
||||
}
|
||||
|
||||
// removeRetired removes every retired consumer, when the connection can.
|
||||
func removeRetired(e Ensurer) error {
|
||||
r, ok := e.(Remover)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
for _, c := range RetiredConsumers {
|
||||
if err := r.RemoveConsumer(c.Stream, c.Name); err != nil {
|
||||
return fmt.Errorf("removing the retired consumer %s on %s: %w", c.Name, c.Stream, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ControllerSeat is the role the control plane holds, and ControllerStates are the facts it states
|
||||
// under it (novox/hq ADR 0134).
|
||||
//
|
||||
@@ -513,23 +536,6 @@ func ControllerConsumers() []Consumer {
|
||||
Why: "the controller is the single consumer of what nodes say; explicit ack and no " +
|
||||
"max-deliver, because the store window's bound is the controller's own",
|
||||
},
|
||||
// The forge's merges (novox/hq ADR 0407 decision 3, issue 973): never dropped, so no max-deliver and no
|
||||
// reset; a merge the controller cannot record yet is held with a growing delay until it can. **Asserted before
|
||||
// the events consumer**, which stops filtering merges: the order of this list is the order of assertion.
|
||||
{
|
||||
Name: MergesConsumerName,
|
||||
Stream: "EVENTS",
|
||||
// Within what the controller may subscribe already (`_DELIVER.controller.>`).
|
||||
DeliverSubject: "_DELIVER." + ControllerName + ".merges",
|
||||
Filters: []string{ControllerFollows[3]},
|
||||
Push: true,
|
||||
AckWaitSeconds: 30,
|
||||
MaxAckPending: 1,
|
||||
FromNow: true,
|
||||
TakesOverFrom: ControllerName,
|
||||
Why: "the forge's merges, one at a time; no max-deliver and never reset, because a merge is never " +
|
||||
"dropped and no person releases one",
|
||||
},
|
||||
{
|
||||
Name: ControllerName,
|
||||
Stream: "EVENTS",
|
||||
@@ -581,6 +587,9 @@ func NoticesConsumer() Consumer {
|
||||
// one on DEAD_LETTER_NOTICES, which the keeper of dead letters asserts and retries by itself, so a fault
|
||||
// there never stops the controller serving (novox/hq issue 330).
|
||||
func AssertServingConsumers(e Ensurer) error {
|
||||
if err := removeRetired(e); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, c := range MeshConsumers() {
|
||||
if c.Stream == DeadLetterNoticesStream {
|
||||
continue
|
||||
@@ -603,6 +612,9 @@ type Ensurer interface {
|
||||
// After the streams, necessarily: a consumer on a stream that does not exist is refused, and the
|
||||
// refusal names the stream rather than the order.
|
||||
func AssertMeshConsumers(e Ensurer) error {
|
||||
if err := removeRetired(e); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, c := range MeshConsumers() {
|
||||
if err := e.EnsureConsumer(c); err != nil {
|
||||
return fmt.Errorf("asserting consumer %s on %s: %w", c.Name, c.Stream, err)
|
||||
|
||||
@@ -30,9 +30,8 @@ var Contracts = map[string]Contract{
|
||||
KindBuilt: {Unordered: "a build's outcome on the retired build seat's subjects: dropped, nothing takes a build in " +
|
||||
"from the bus any more (R4g, ADR 0501 §1); the subjects stay followed while the installer's carried user list " +
|
||||
"names them"},
|
||||
KindSourceMoved: {Ordered: "by the merge's commit against what was built from it: a merge already acted on " +
|
||||
"or older than the last look is history, not a second plan (issues 250, 266)",
|
||||
Tests: []string{"TestAMergeOlderThanTheLastLookIsHistory", "TestAMergeMatchesTheSourcesBuiltFromIt"}},
|
||||
KindSourceMoved: {Unordered: "the forge's merge on EVENTS: no longer followed, dropped if one arrives; the forge " +
|
||||
"bridge's stream FORGE is the record of the merges, read at a moved module's registration (sweep 2026-10-30)"},
|
||||
KindToolsHeartbeat: {Unordered: "a word that the node tools are there, as a machine's heartbeat"},
|
||||
KindHealth: {Ordered: "by the time the node-engine looked (`at`, on the machine's clock): a statement older " +
|
||||
"than the one kept for that machine is refused, so an event arriving after a newer report does not undo it " +
|
||||
|
||||
@@ -62,7 +62,7 @@ func Nats(js *broker.JetStream) Inbound {
|
||||
// whatever was asked for — and not at all when nothing was.
|
||||
func (n *natsInbound) Also(kind string) error {
|
||||
switch kind {
|
||||
case KindModuleMoved, KindCatchUp, KindSourceMoved, KindProvisioner, KindPullUpdated, KindDecided,
|
||||
case KindModuleMoved, KindCatchUp, KindProvisioner, KindPullUpdated, KindDecided,
|
||||
KindDeliveryMerged:
|
||||
n.follows[kind] = true
|
||||
return nil
|
||||
@@ -145,24 +145,6 @@ func (n *natsInbound) Receive(ctx context.Context, act func(context.Context, Con
|
||||
defer func() { _ = followed.Unsubscribe() }()
|
||||
}
|
||||
|
||||
// And the forge's merges, off a consumer of their own: no max-deliver and no reset, so a merge held for the
|
||||
// plans waits for them however long (novox/hq ADR 0407 decision 3). Bound to the same channel: one loop.
|
||||
if n.follows[KindSourceMoved] {
|
||||
if events == nil {
|
||||
events = make(chan *nats.Msg, Prefetch)
|
||||
}
|
||||
merges, err := standingBy(ctx, log.Default(), "EVENTS", func() (*nats.Subscription, error) {
|
||||
return js.ChanSubscribe(MergedSubject, events, nats.Bind("EVENTS", broker.MergesConsumerName))
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("subscribing to the forge's merges: %w", err)
|
||||
}
|
||||
if merges == nil {
|
||||
return nil
|
||||
}
|
||||
defer func() { _ = merges.Unsubscribe() }()
|
||||
}
|
||||
|
||||
// A connection that dropped is said, not discovered. A controller whose bus connection is gone
|
||||
// is a mesh where nothing can be told anything.
|
||||
gone := make(chan error, 1)
|
||||
@@ -423,7 +405,11 @@ func (m *natsControl) Hold(after time.Duration) error {
|
||||
if _, already := m.on.since[m.seq]; !already {
|
||||
m.on.since[m.seq] = time.Now()
|
||||
}
|
||||
return m.msg.NakWithDelay(after)
|
||||
if err := m.msg.NakWithDelay(after); err != nil {
|
||||
return err
|
||||
}
|
||||
Loop.HeldUntil(time.Now().Add(after))
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *natsControl) forget() {
|
||||
|
||||
@@ -394,35 +394,3 @@ func TestNatsAReportThatIsNotNewsIsNotStated(t *testing.T) {
|
||||
case <-time.After(3 * time.Second):
|
||||
}
|
||||
}
|
||||
|
||||
func TestNatsAMergeHeldMoreThanFiveTimesIsStillHandedOverAndRecorded(t *testing.T) {
|
||||
first, most := MergeHeldFirst, MergeHeldMost
|
||||
MergeHeldFirst, MergeHeldMost = 5*time.Millisecond, 40*time.Millisecond
|
||||
t.Cleanup(func() { MergeHeldFirst, MergeHeldMost = first, most })
|
||||
js := aBus(t)
|
||||
up := &manyTimesBusy{busy: 8}
|
||||
s := &Server{inbound: Nats(js), bus: OverNATS{Conn: js.Conn(), JS: js.Context()}, log: quiet()}
|
||||
if err := s.Follows(up); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, stop := context.WithCancel(context.Background())
|
||||
defer stop()
|
||||
go func() { _ = s.Serve(ctx) }()
|
||||
eventually(t, "the merges consumer being made", func() bool {
|
||||
_, err := js.Context().ConsumerInfo("EVENTS", broker.MergesConsumerName)
|
||||
return err == nil
|
||||
})
|
||||
|
||||
body, _ := json.Marshal(SourceMoved{Owner: "novox", Repo: "one", Base: "main", Commit: "abcdef0123"})
|
||||
if _, err := js.Context().Publish(MergedSubject, body); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
eventually(t, "the ninth hand-over recording the merge", func() bool { return up.called() == 9 })
|
||||
eventually(t, "the merge being acknowledged", func() bool {
|
||||
info, err := js.Context().ConsumerInfo("EVENTS", broker.MergesConsumerName)
|
||||
return err == nil && info.NumAckPending == 0 && info.NumPending == 0
|
||||
})
|
||||
if info, err := js.Context().StreamInfo(broker.DeadLettersStream); err == nil && info.State.Msgs != 0 {
|
||||
t.Fatalf("%d message(s) in dead letters: a merge was given up on", info.State.Msgs)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,7 +2,6 @@ package link
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"log"
|
||||
"sync"
|
||||
@@ -199,46 +198,3 @@ func (s slowly) Heard(context.Context, Report) (bool, error) { s.work(); return
|
||||
type sameAgain struct{}
|
||||
|
||||
func (sameAgain) Heard(context.Context, Report) (bool, error) { return false, nil }
|
||||
|
||||
// manyTimesBusy is an upgrader whose plans are busy for its first merges, as another controller's hold is.
|
||||
type manyTimesBusy struct {
|
||||
toldAbout
|
||||
busy int
|
||||
calls int
|
||||
}
|
||||
|
||||
func (u *manyTimesBusy) SourceMoved(context.Context, SourceMoved) error {
|
||||
u.mu.Lock()
|
||||
defer u.mu.Unlock()
|
||||
u.calls++
|
||||
if u.calls <= u.busy {
|
||||
return errors.Join(ErrTryAgain, errors.New("another controller is working the plans"))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (u *manyTimesBusy) called() int {
|
||||
u.mu.Lock()
|
||||
defer u.mu.Unlock()
|
||||
return u.calls
|
||||
}
|
||||
|
||||
// **A merge held more than five times is still handed over and recorded** (issue 973, novox/hq ADR 0407 decision
|
||||
// 3): the merges have a durable consumer of their own with no max-deliver, so they never go to DEAD_LETTERS to wait
|
||||
// for a person; the delay grows, and is capped.
|
||||
|
||||
// The delay a held merge waits grows from the first to the most and stays there.
|
||||
func TestAHeldMergesDelayGrowsAndIsCapped(t *testing.T) {
|
||||
var got []time.Duration
|
||||
for _, held := range []time.Duration{0, 4 * time.Second, 10 * time.Second, 25 * time.Second, 70 * time.Second,
|
||||
200 * time.Second, 20 * time.Minute} {
|
||||
got = append(got, MergeRedeliverAfter(held))
|
||||
}
|
||||
want := []time.Duration{5 * time.Second, 5 * time.Second, 10 * time.Second, 25 * time.Second, 70 * time.Second,
|
||||
200 * time.Second, 5 * time.Minute}
|
||||
for i := range want {
|
||||
if got[i] != want[i] {
|
||||
t.Errorf("held %d: waits %v; want %v", i, got[i], want[i])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+3
-61
@@ -49,8 +49,6 @@ type Upgrader interface {
|
||||
// stop every upgrade behind it — except the store unreachable for the moment, which is asked
|
||||
// again for a bounded time (novox/hq issue 083).
|
||||
Upgraded(ctx context.Context, u Upgraded) error
|
||||
// SourceMoved is a merge on the forge: build what that source produces, bases first.
|
||||
SourceMoved(ctx context.Context, m SourceMoved) error
|
||||
}
|
||||
|
||||
// Checker is what the controller does when the forge says a pull request's head moved: ask for it to be
|
||||
@@ -105,9 +103,6 @@ const GiveUpAfter = 2 * time.Minute
|
||||
|
||||
// Follows says what to do about upgrades, and asks for them to be delivered.
|
||||
func (s *Server) Follows(u Upgrader) error {
|
||||
if err := s.inbound.Also(KindSourceMoved); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := s.inbound.Also(KindModuleMoved); err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -225,7 +220,9 @@ func (s *Server) act(ctx context.Context, m Control) {
|
||||
case KindModuleMoved:
|
||||
s.moved(ctx, m)
|
||||
case KindSourceMoved:
|
||||
s.sourceMoved(ctx, m)
|
||||
// The forge's merges are no longer followed (sweep 2026-10-30): the forge bridge's stream is their record, read
|
||||
// where a merge is needed (keptForgeMerge). One that arrives on an old consumer is let go of.
|
||||
_ = m.Drop()
|
||||
case KindCatchUp:
|
||||
s.catchingUp(ctx, m)
|
||||
case KindProvisioner:
|
||||
@@ -521,61 +518,6 @@ func short(commit string) string {
|
||||
return commit
|
||||
}
|
||||
|
||||
// MergeHeldFirst is how long a merge the controller could not record yet waits before the bus hands it over
|
||||
// again, and MergeHeldMost the longest it ever waits: the delay grows with how long the merge has been held, from
|
||||
// the first to the most (novox/hq ADR 0407 decision 3). Variables so a test need not wait.
|
||||
var (
|
||||
MergeHeldFirst = 5 * time.Second
|
||||
MergeHeldMost = 5 * time.Minute
|
||||
)
|
||||
|
||||
// MergeRedeliverAfter is the delay for a merge held for the time given: that time, so the delay about doubles
|
||||
// with each hold, between MergeHeldFirst and MergeHeldMost.
|
||||
func MergeRedeliverAfter(heldFor time.Duration) time.Duration {
|
||||
return min(max(heldFor, MergeHeldFirst), MergeHeldMost)
|
||||
}
|
||||
|
||||
// sourceMoved acts on the forge's announcement of a merge. **A merge is never dropped** (issue 921, novox/hq ADR
|
||||
// 0407 decision 3): one the controller could not record for now (the plans held by another controller, the store
|
||||
// away: ErrTryAgain) is held and handed over again until it is recorded; the controller giving up on it would
|
||||
// lose a change for good. Anything else is an answer and is taken: a build that fails is reported by the build
|
||||
// itself, and re-delivering the merge would only re-fail it.
|
||||
func (s *Server) sourceMoved(ctx context.Context, m Control) {
|
||||
var moved SourceMoved
|
||||
if err := json.Unmarshal(m.Body(), &moved); err != nil {
|
||||
s.log.Printf("a merge announcement could not be read: %v", err)
|
||||
_ = m.Took()
|
||||
return
|
||||
}
|
||||
if moved.Commit == "" || moved.Repo == "" {
|
||||
s.log.Printf("a merge announcement named no repository or no commit; ignored")
|
||||
_ = m.Took()
|
||||
return
|
||||
}
|
||||
err := s.upgrader.SourceMoved(ctx, moved)
|
||||
// Left unsettled when the controller is ending, whatever error that gave (the plans' hold returns a bare
|
||||
// context.Canceled): the bus hands it to whatever consumes next, and taking it would lose the merge.
|
||||
if err != nil && ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
if err != nil && errors.Is(err, ErrTryAgain) {
|
||||
// Also left unsettled when the bus will not hold it.
|
||||
delay := MergeRedeliverAfter(m.HeldFor())
|
||||
if held := m.Hold(delay); held != nil {
|
||||
s.log.Printf("%s/%s moved to %.8s and could not be held (%v): %v; left for the bus to hand over again",
|
||||
moved.Owner, moved.Repo, moved.Commit, held, err)
|
||||
return
|
||||
}
|
||||
s.log.Printf("%s/%s moved to %.8s and is held: %v; handed over again in %s", moved.Owner, moved.Repo,
|
||||
moved.Commit, err, delay)
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
s.log.Printf("%s/%s moved to %.8s and the mesh could not act on it: %v", moved.Owner, moved.Repo, moved.Commit, err)
|
||||
}
|
||||
_ = m.Took()
|
||||
}
|
||||
|
||||
// pullUpdated asks for a pull request's merge check. Taken whatever happens: a check that could not be
|
||||
// asked is said here, and the next push to the pull request asks again.
|
||||
func (s *Server) pullUpdated(ctx context.Context, m Control) {
|
||||
|
||||
@@ -1,60 +0,0 @@
|
||||
package link
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
"log"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// heldControl is a merge announcement that records how it was settled.
|
||||
type heldControl struct {
|
||||
body []byte
|
||||
took, held, drops int
|
||||
}
|
||||
|
||||
func (c *heldControl) Kind() string { return "source-moved" }
|
||||
func (c *heldControl) Body() []byte { return c.body }
|
||||
func (c *heldControl) Subject() string { return "" }
|
||||
func (c *heldControl) Redelivered() bool { return false }
|
||||
func (c *heldControl) HeldFor() time.Duration { return 0 }
|
||||
func (c *heldControl) Answer(context.Context, []byte) error { return nil }
|
||||
func (c *heldControl) Took() error { c.took++; return nil }
|
||||
func (c *heldControl) Hold(time.Duration) error { c.held++; return nil }
|
||||
func (c *heldControl) Drop() error { c.drops++; return nil }
|
||||
|
||||
type sourceMover struct {
|
||||
Upgrader
|
||||
answer error
|
||||
}
|
||||
|
||||
func (s sourceMover) SourceMoved(context.Context, SourceMoved) error { return s.answer }
|
||||
|
||||
// **A merge is acknowledged only after the controller recorded it** (novox/hq ADR 0407 decision 3, issue 1092): an
|
||||
// answer of "not now" — the plan it belongs to could not be saved, the bus away, the lease lost — leaves the announcement
|
||||
// unacknowledged, held to be handed over again; only an answer that is not "not now" settles it.
|
||||
func TestAMergeTheControllerCouldNotRecordIsHeldAndNotAcknowledged(t *testing.T) {
|
||||
body, err := json.Marshal(SourceMoved{Owner: "novox", Repo: "mesh-catalog", Base: "main", Commit: "c1"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
s := &Server{log: log.New(io.Discard, "", 0)}
|
||||
for name, tc := range map[string]struct {
|
||||
answer error
|
||||
took, held int
|
||||
}{
|
||||
"recorded": {nil, 1, 0},
|
||||
"not now": {errors.Join(ErrTryAgain, errors.New("the plan is not saved")), 0, 1},
|
||||
"an answer, not a wait": {errors.New("the mesh could not act on it"), 1, 0},
|
||||
} {
|
||||
s.upgrader = sourceMover{answer: tc.answer}
|
||||
m := &heldControl{body: body}
|
||||
s.sourceMoved(context.Background(), m)
|
||||
if m.took != tc.took || m.held != tc.held || m.drops != 0 {
|
||||
t.Errorf("%s: acknowledged %d times and held %d times, want %d and %d", name, m.took, m.held, tc.took, tc.held)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -17,8 +17,7 @@ func (r recordsWith) Built(context.Context, BuildResult) error { return r.err }
|
||||
|
||||
type upgradesWith struct{ err error }
|
||||
|
||||
func (u upgradesWith) Upgraded(context.Context, Upgraded) error { return u.err }
|
||||
func (u upgradesWith) SourceMoved(context.Context, SourceMoved) error { return u.err }
|
||||
func (u upgradesWith) Upgraded(context.Context, Upgraded) error { return u.err }
|
||||
|
||||
type replaysWith struct{ err error }
|
||||
|
||||
@@ -69,61 +68,3 @@ func TestACatchUpWaitsOutARestartingStore(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A merge the controller could not record for now (the plans held by another controller, the store away) is held
|
||||
// and handed over again, never taken: the bus keeps it until it is recorded (issue 921). Any other failure is an
|
||||
// answer and is taken, as before.
|
||||
func TestAMergeThatCouldNotBeRecordedNowIsHeldNotDropped(t *testing.T) {
|
||||
merged := SourceMoved{Owner: "novox", Repo: "one", Base: "main", Commit: "abcdef0123"}
|
||||
for _, c := range []struct {
|
||||
what string
|
||||
err error
|
||||
want func(*settled) bool
|
||||
}{
|
||||
{"the plans held by another controller", errors.Join(ErrTryAgain, errors.New("another controller is working the plans")),
|
||||
func(s *settled) bool { return s.unsettled() }},
|
||||
{"the store away", errors.Join(ErrTryAgain, restarting), func(s *settled) bool { return s.unsettled() }},
|
||||
{"cannot act", errors.New("no such module"), func(s *settled) bool { return s.acked && !s.nacked }},
|
||||
{"kept", nil, func(s *settled) bool { return s.acked && !s.nacked }},
|
||||
} {
|
||||
s, in := serving()
|
||||
s.upgrader = upgradesWith{err: c.err}
|
||||
to := &settled{}
|
||||
held := in.sends(t, to, KindSourceMoved, merged).(*fakeControl)
|
||||
s.act(context.Background(), held)
|
||||
if !c.want(to) {
|
||||
t.Errorf("%s: a merge was settled as %+v", c.what, *to)
|
||||
}
|
||||
if c.err != nil && errors.Is(c.err, ErrTryAgain) && len(in.held) != 1 {
|
||||
t.Errorf("%s: the merge was not held to be handed over again", c.what)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A merge being held when the controller is replaced (its context ended) or when the bus will not hold it is left
|
||||
// unsettled, never taken: the bus hands it to whatever consumes next (review of issue 921).
|
||||
func TestAMergeIsLeftUnsettledWhenTheControllerEndsOrTheHoldFails(t *testing.T) {
|
||||
merged := SourceMoved{Owner: "novox", Repo: "one", Base: "main", Commit: "abcdef0123"}
|
||||
try := errors.Join(ErrTryAgain, errors.New("another controller is working the plans"))
|
||||
|
||||
// The plans' hold returns a bare context.Canceled when the controller is replaced, not ErrTryAgain.
|
||||
s, in := serving()
|
||||
s.upgrader = upgradesWith{err: context.Canceled}
|
||||
to := &settled{}
|
||||
ended, end := context.WithCancel(context.Background())
|
||||
end()
|
||||
s.act(ended, in.sends(t, to, KindSourceMoved, merged))
|
||||
if !to.unsettled() {
|
||||
t.Errorf("a merge was settled as %+v when the controller's context ended", *to)
|
||||
}
|
||||
|
||||
s, in = serving()
|
||||
s.upgrader = upgradesWith{err: try}
|
||||
to = &settled{}
|
||||
m := in.sends(t, to, KindSourceMoved, merged).(*fakeControl)
|
||||
m.holdFails = errors.New("the bus will not hold it")
|
||||
s.act(context.Background(), m)
|
||||
if !to.unsettled() {
|
||||
t.Errorf("a merge was settled as %+v when it could not be held", *to)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -85,6 +85,26 @@ func nodeOfToolsAlive(subject string) (string, bool) {
|
||||
type LoopActivity struct {
|
||||
mu sync.Mutex
|
||||
took time.Time
|
||||
// due is when the latest message the loop held back with a delay is handed over again: until then the loop waits
|
||||
// by its own choice, which is not deafness (sweep 2026-10-30: a merge held five minutes at a time read as a deaf
|
||||
// controller every time its delay outran the deaf bound).
|
||||
due time.Time
|
||||
}
|
||||
|
||||
// HeldUntil records that the loop handed a message back to be delivered again at due.
|
||||
func (l *LoopActivity) HeldUntil(due time.Time) {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
if due.After(l.due) {
|
||||
l.due = due
|
||||
}
|
||||
}
|
||||
|
||||
// Due is when the latest message the loop held is handed over again; zero when it held none.
|
||||
func (l *LoopActivity) Due() time.Time {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
return l.due
|
||||
}
|
||||
|
||||
// Loop is this process's event loop.
|
||||
|
||||
Reference in New Issue
Block a user