package link import ( "context" "crypto/ed25519" "encoding/json" "errors" "reflect" "sync" "testing" "time" ) // A report of an apply reaches the mesh even when the apply ended the link, and one lost anyway is // said again the next time the node is linked (novox/hq issue 264). // quietLink is a link with no broker behind it. A report is refused once the context it is // published on is cancelled, as the bus's is ("reporting: context canceled"), or while refusing. type quietLink struct { mu sync.Mutex reports []Report refusing bool delivered chan Declaration lost chan error } func newQuietLink(arriving ...Declaration) *quietLink { l := &quietLink{delivered: make(chan Declaration, len(arriving)), lost: make(chan error, 1)} for _, d := range arriving { l.delivered <- d } return l } func (l *quietLink) Report(ctx context.Context, _ string, body []byte) error { if err := ctx.Err(); err != nil { return err } l.mu.Lock() defer l.mu.Unlock() if l.refusing { return errors.New("refused") } var r Report if err := json.Unmarshal(body, &r); err != nil { return err } l.reports = append(l.reports, r) return nil } func (l *quietLink) Alive(context.Context, string, []byte) error { return nil } func (l *quietLink) Declarations() <-chan Declaration { return l.delivered } func (l *quietLink) Lost() <-chan error { return l.lost } func (l *quietLink) Close() {} func (l *quietLink) said() []Report { l.mu.Lock() defer l.mu.Unlock() return append([]Report(nil), l.reports...) } // keptInMemory is Unsaid without a disk. type keptInMemory struct { report *Report } func (k *keptInMemory) Keep(r Report) error { if r.Declared == "" { k.report = nil return nil } k.report = &r return nil } func (k *keptInMemory) Said(declared string) error { if k.report != nil && k.report.Declared == declared { k.report = nil } return nil } func (k *keptInMemory) Pending() (Report, bool, error) { if k.report == nil { return Report{}, false, nil } return *k.report, true, nil } func aMember(t *testing.T) (Membership, ed25519.PrivateKey) { t.Helper() public, private, err := ed25519.GenerateKey(nil) if err != nil { t.Fatal(err) } return Membership{Node: "n1", Signer: public}, private } // Measured on the anchor: a declaration carried a new controller and a new host, the host applied // it, stood aside for its successor — which cancels the link — and the report of that apply failed // with "context canceled". The plan waited on it until somebody pushed by hand. func TestAnApplyThatEndsTheLinkIsStillReported(t *testing.T) { m, key := aMember(t) declaration := &said{body: signedBy(t, key, []byte(`{"declaration":1}`))} l := newQuietLink(declaration) kept := &keptInMemory{} ctx, standAside := context.WithCancel(context.Background()) defer standAside() apply := func(context.Context, []byte, []byte) Report { standAside() // the host delivered its successor, and stands aside for it return Report{Declared: "d1", Applied: []string{"a"}} } done := make(chan error, 1) go func() { done <- serve(ctx, l, m, apply, nil, time.Second, nil, kept) }() select { case err := <-done: if err != nil { t.Fatal(err) } case <-time.After(5 * time.Second): t.Fatal("the link did not end after the apply stood aside") } reports := l.said() if len(reports) != 1 || reports[0].Declared != "d1" || !reflect.DeepEqual(reports[0].Applied, []string{"a"}) { t.Fatalf("the apply's report did not reach the mesh: %+v", reports) } if !declaration.handled { t.Fatal("the declaration was not settled after its report") } if _, ok, _ := kept.Pending(); ok { t.Fatal("a report the mesh took is still kept to be said again") } } // A report that did not reach the mesh — the host died between applying and publishing, or the // broker refused it — is said on the next link, before anything else, and exactly as it was made. func TestAReportLostAfterTheApplyIsSaidOnTheNextLink(t *testing.T) { m, key := aMember(t) made := Report{Declared: "d1", Applied: []string{"a"}, Failed: map[string]string{"b": "no room"}} kept := &keptInMemory{} // The apply happens and its report is lost. first := newQuietLink(&said{body: signedBy(t, key, []byte(`{"declaration":1}`))}) first.refusing = true ctx, stop := context.WithCancel(context.Background()) apply := func(context.Context, []byte, []byte) Report { stop(); return made } if err := serve(ctx, first, m, apply, nil, time.Second, nil, kept); err != nil { t.Fatal(err) } if len(first.said()) != 0 { t.Fatal("the broker refused the report and it counted as said") } // The next host links. It is stopped at once, so all it does is what it does on linking. next := newQuietLink() gone, cancel := context.WithCancel(context.Background()) cancel() never := func(context.Context, []byte, []byte) Report { t.Fatal("nothing was delivered, and something was applied") return Report{} } if err := serve(gone, next, m, never, nil, time.Second, nil, kept); err != nil { t.Fatal(err) } reports := next.said() made.Node = m.Node if len(reports) != 1 || !reflect.DeepEqual(reports[0], made) { t.Fatalf("the lost report was not said again as it was made: %+v", reports) } if _, ok, _ := kept.Pending(); ok { t.Fatal("a report said again is still kept") } } // A node that never applied anything, or whose last report was taken, says nothing on linking. func TestNothingIsSaidAgainWhenNothingWasLost(t *testing.T) { m, _ := aMember(t) l := newQuietLink() gone, cancel := context.WithCancel(context.Background()) cancel() if err := serve(gone, l, m, nil, nil, time.Second, nil, &keptInMemory{}); err != nil { t.Fatal(err) } if err := serve(gone, l, m, nil, nil, time.Second, nil, nil); err != nil { t.Fatal(err) } if reports := l.said(); len(reports) != 0 { t.Fatalf("a node with nothing lost reported %+v", reports) } }