jschoubben
1ebad3786c
The pipeline was observable from a merge to an artifact and went dark where it touched a machine: a node's report is control traffic only the control plane reads, so nothing said which version a machine runs, or that it refused to (novox/hq ADR 0134). The control plane now states both under the seat it holds — a role's events belong to the role and keep their address when the holder is replaced — and only when the report is news, because a machine reconciles every minute and a fact per report would be a fact per minute per machine. Whether a report is news is the store's answer: it holds the previous one, so the listener returns it and the server states the fact. That also gives the catch-up replay a subject the controller may publish: it was published as a module's event from a module called "control-plane", which does not exist, so the controller's own account refused it and every catalogue that asked what it missed was answered with nothing.
66 lines
2.4 KiB
Go
66 lines
2.4 KiB
Go
package link
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
type heardWith struct{ err error }
|
|
|
|
func (h heardWith) Heard(context.Context, Report) (bool, error) { return h.err == nil, h.err }
|
|
|
|
// switchable answers with whatever it is set to — the store away, then back.
|
|
type switchable struct{ err error }
|
|
|
|
func (h *switchable) Heard(context.Context, Report) (bool, error) { return h.err == nil, h.err }
|
|
|
|
func aReport(node, declared string) Report {
|
|
return Report{Node: node, Declared: declared, Applied: []string{"store"}}
|
|
}
|
|
|
|
// A report the store could not take right now is held, unsettled, and recorded when the store is
|
|
// back; one the store answered no to is acknowledged; one recorded is acknowledged (issue 082, 083).
|
|
func TestAReportTheStoreCouldNotTakeIsHeldAndOneItRefusedIsNot(t *testing.T) {
|
|
store := &switchable{err: errors.Join(ErrTryAgain, errors.New("starting up"))}
|
|
s, in := serving()
|
|
s.listener = store
|
|
held := &settled{}
|
|
s.act(context.Background(), in.sends(t, held, KindReport, aReport("anchor", "d1")))
|
|
if !held.unsettled() || len(in.held) != 1 {
|
|
t.Fatalf("a report the store could not take was not held: %+v, %d held", held, len(in.held))
|
|
}
|
|
store.err = nil
|
|
in.retries(context.Background(), s)
|
|
if !held.acked || len(in.held) != 0 {
|
|
t.Fatalf("a held report was not recorded once the store was back: %+v, %d held", held, len(in.held))
|
|
}
|
|
|
|
refused := &settled{}
|
|
s, in = serving()
|
|
s.listener = heardWith{err: errors.New("a report named no node")}
|
|
s.act(context.Background(), in.sends(t, refused, KindReport, aReport("anchor", "d1")))
|
|
if !refused.acked || refused.nacked {
|
|
t.Fatalf("a report the store answered no to was not acknowledged: %+v", refused)
|
|
}
|
|
}
|
|
|
|
// A store that has not come back within the bound is not restarting: the report is let go, loudly,
|
|
// rather than held for ever.
|
|
func TestAReportIsLetGoOnceTheStoreHasBeenGoneTooLong(t *testing.T) {
|
|
s, in := serving()
|
|
s.listener = heardWith{err: errors.Join(ErrTryAgain, errors.New("connection refused"))}
|
|
s.giveUp = time.Millisecond
|
|
held := &settled{}
|
|
s.act(context.Background(), in.sends(t, held, KindReport, aReport("anchor", "d1")))
|
|
if !held.unsettled() {
|
|
t.Fatalf("the first failure was not held: %+v", held)
|
|
}
|
|
time.Sleep(5 * time.Millisecond)
|
|
in.retries(context.Background(), s)
|
|
if !held.acked || len(in.held) != 0 {
|
|
t.Fatalf("a report past the bound was not let go: %+v, %d held", held, len(in.held))
|
|
}
|
|
}
|