diff --git a/cmd/mesh-controller/delivery.go b/cmd/mesh-controller/delivery.go index 615386c0..40b7c489 100644 --- a/cmd/mesh-controller/delivery.go +++ b/cmd/mesh-controller/delivery.go @@ -1042,13 +1042,13 @@ func deliveryCommand(ctx context.Context, args []string) error { return fmt.Errorf("delivery %s: plan, order, check, go, stop or walks", sub) } -// appliedFrom answers a machine's first report after a send from the controller's `apply` durations; a lookup -// that fails is a report not read, which leaves the walk's end unknown rather than wrong. +// appliedFrom answers a machine's own report of the send a walk made to it from the controller's `apply` durations +// (novox/hq issue 582); a lookup that fails is a report not read, which leaves the walk's end unknown rather than wrong. func appliedFrom(ctx context.Context, inv *inventory.Inventory) inventory.AppliedLookup { - return func(node string, sent time.Time) (inventory.AppliedReport, bool) { - r, ok, err := inv.FirstAppliedAfter(ctx, node, sent) + return func(node string, send inventory.SentDeclaration, walkSent time.Time) (inventory.AppliedReport, bool) { + r, ok, err := inv.FirstAppliedAfter(ctx, node, send, walkSent) if err != nil { - fmt.Fprintf(os.Stderr, "the report of %s after %s could not be read: %v\n", node, sent.Format(time.RFC3339), err) + fmt.Fprintf(os.Stderr, "the report of %s after %s could not be read: %v\n", node, walkSent.Format(time.RFC3339), err) return inventory.AppliedReport{}, false } return r, ok diff --git a/internal/inventory/nodes.go b/internal/inventory/nodes.go index 4c88e0e3..8552d1ff 100644 --- a/internal/inventory/nodes.go +++ b/internal/inventory/nodes.go @@ -1064,9 +1064,14 @@ type Reported struct { // SentDeclaration is what one send carried to a machine, as a gate keeps it: the declaration's digest and // its sequence (novox/hq issue 352). A report about this declaration, or about one sequenced after it, is a // report on what the gate sent — whatever the machine was sent since. +// +// At is when the controller sent it to that machine: the node's own send time, which is where that machine's +// `apply` duration starts. A walk measures its rest from it, not from its own record of the send, made only after +// every machine of the rest was sent (novox/hq issue 582). Zero for a send kept before it was recorded. type SentDeclaration struct { - Digest string `json:"digest"` - Sequence int64 `json:"sequence,omitempty"` + Digest string `json:"digest"` + Sequence int64 `json:"sequence,omitempty"` + At time.Time `json:"at,omitzero"` } // ReportsOn says a report is about this send: the declaration itself; one the same machine was sequenced @@ -1085,7 +1090,8 @@ func (s SentDeclaration) ReportsOn(r Reported) bool { func (i *Inventory) SentTo(ctx context.Context, name string) (SentDeclaration, bool, error) { var digest *string var seq *int64 - err := i.store.Pool().QueryRow(ctx, `select sent, sequence from node where name = $1`, name).Scan(&digest, &seq) + var at *time.Time + err := i.store.Pool().QueryRow(ctx, `select sent, sequence, sent_at from node where name = $1`, name).Scan(&digest, &seq, &at) if errors.Is(err, pgx.ErrNoRows) { return SentDeclaration{}, false, fmt.Errorf("%w: %s", ErrNoSuchNode, name) } @@ -1096,6 +1102,9 @@ func (i *Inventory) SentTo(ctx context.Context, name string) (SentDeclaration, b if seq != nil { s.Sequence = *seq } + if at != nil { + s.At = at.UTC() + } return s, true, nil } diff --git a/internal/inventory/walkphases.go b/internal/inventory/walkphases.go index 23f1eeb7..971945c3 100644 --- a/internal/inventory/walkphases.go +++ b/internal/inventory/walkphases.go @@ -2,13 +2,10 @@ package inventory import ( "context" - "errors" "fmt" "sort" "strings" "time" - - "github.com/jackc/pgx/v5" ) // Where a walk's time went (novox/hq ADR 0282 decision 6, issue 382): from its merge to every machine of its @@ -59,11 +56,14 @@ type WalkPhases struct { TotalMS int64 `json:"total_ms,omitempty"` Total string `json:"total,omitempty"` // UnknownMS is the time within the walk no phase could be measured over. - UnknownMS int64 `json:"unknown_ms,omitempty"` - Phases []WalkPhase `json:"phases"` - Silent []string `json:"silent,omitempty"` - Said string `json:"said"` - Merges []MergeStart `json:"merges,omitempty"` + UnknownMS int64 `json:"unknown_ms,omitempty"` + Phases []WalkPhase `json:"phases"` + Silent []string `json:"silent,omitempty"` + // Rest is each machine of the rest's own report of the send that carried the walk's build there (novox/hq ADR + // 0282 decision 6, issue 582): when it was sent, when it said, and what. + Rest []RestReport `json:"rest,omitempty"` + Said string `json:"said"` + Merges []MergeStart `json:"merges,omitempty"` } // MergeStart is one merge the walk answers and when it was made: where that merge's delivery time starts. @@ -92,8 +92,34 @@ type AppliedReport struct { Outcome string } -// AppliedLookup answers a machine's first report after a send to it; false when it has not reported. -type AppliedLookup func(node string, sent time.Time) (AppliedReport, bool) +// AppliedLookup answers a machine's report of the send the walk made to it: send is what the walk kept of it, and +// walkSent the walk's own record of its send to the rest; false when it has not reported. +type AppliedLookup func(node string, send SentDeclaration, walkSent time.Time) (AppliedReport, bool) + +// RestReport is one machine of the rest and its own report of a send of the walk (novox/hq issue 582). +type RestReport struct { + Node string `json:"node"` + Modules []string `json:"modules"` + // Sent is when that machine was sent, by its own send time where the walk kept it; else the walk's record of + // its send to the rest, and SentSaid says so. + Sent time.Time `json:"sent"` + SentSaid string `json:"sent_said,omitempty"` + Sequence int64 `json:"sequence,omitempty"` + Reported *time.Time `json:"reported,omitempty"` + Took string `json:"took,omitempty"` + Outcome string `json:"outcome,omitempty"` + // State is applied, failed, waiting (not yet heard, within ApplySilentAfter) or silent (not heard within it: + // listed, never moving the walk's end). + State string `json:"state"` +} + +// The states of a machine of the rest. +const ( + RestApplied = "applied" + RestFailed = "failed" + RestWaiting = "waiting" + RestSilent = "silent" +) // point is one moment of the walk, ending the phase named. type point struct { @@ -307,8 +333,8 @@ func (p Plan) points(now time.Time, applied AppliedLookup, w *WalkPhases) []poin } } else { rest = latest(rest, s.SentAt) - for node := range s.Rest { - lastSends = append(lastSends, restSend{module: m, node: node, at: *s.SentAt}) + for node, what := range s.Rest { + lastSends = append(lastSends, restSend{module: m, node: node, walkSent: *s.SentAt, what: what}) } } } @@ -347,22 +373,73 @@ func (p Plan) points(now time.Time, applied AppliedLookup, w *WalkPhases) []poin if applied == nil { return append(out, point{phase: PhaseApply, tier: -1, said: "the rest's reports were not read"}) } - var end *time.Time - var waiting, failed []string + // Each machine of the rest, by its own report of the send that carried the build there (novox/hq issue 582): + // never a report of a send made after it, which another walk's send may be. + reports := map[string]*RestReport{} + var order []string for _, s := range lastSends { - r, ok := applied(s.node, s.at) - switch { - case !ok && now.Sub(s.at) > ApplySilentAfter, ok && r.At.Sub(s.at) > ApplySilentAfter: - w.Silent = appendOnce(w.Silent, s.node) - case !ok: - waiting = appendOnce(waiting, s.node) - case r.Outcome != OutcomeApplied: - failed = appendOnce(failed, s.node+" ("+r.Outcome+")") - default: - at := r.At - end = latest(end, &at) + at, said := s.walkSent, "" + if !s.what.At.IsZero() { + at = s.what.At + } else { + said = "the walk's record of its send to the rest: this machine's own send time was not kept" + } + key := s.node + "\x00" + s.what.Digest + "\x00" + at.Format(time.RFC3339Nano) + if r := reports[key]; r != nil { + r.Modules = appendOnce(r.Modules, s.module) + continue + } + rr := &RestReport{Node: s.node, Modules: []string{s.module}, Sent: at, SentSaid: said, Sequence: s.what.Sequence} + if r, ok := applied(s.node, s.what, s.walkSent); ok { + got := r.At + rr.Reported, rr.Outcome, rr.Took = &got, r.Outcome, words(r.At.Sub(at)) + } + reports[key] = rr + order = append(order, key) + } + // A machine sent twice by the walk (one tier's rest, then the next's) that said nothing of the first send has + // said it of the second, which carries the first's builds too: both are this walk's. + for _, k := range order { + rr := reports[k] + if rr.Reported != nil { + continue + } + for _, l := range order { + later := reports[l] + if later.Node == rr.Node && later.Sent.After(rr.Sent) && later.Reported != nil && + (rr.Reported == nil || later.Reported.Before(*rr.Reported)) { + got := *later.Reported + rr.Reported, rr.Outcome, rr.Took = &got, later.Outcome, words(got.Sub(rr.Sent)) + } } } + var end *time.Time + var waiting, failed []string + for _, k := range order { + rr := reports[k] + switch { + case rr.Reported == nil && now.Sub(rr.Sent) > ApplySilentAfter, rr.Reported != nil && rr.Reported.Sub(rr.Sent) > ApplySilentAfter: + rr.State = RestSilent + w.Silent = appendOnce(w.Silent, rr.Node) + case rr.Reported == nil: + rr.State = RestWaiting + waiting = appendOnce(waiting, rr.Node) + case rr.Outcome != OutcomeApplied: + rr.State = RestFailed + failed = appendOnce(failed, rr.Node+" ("+rr.Outcome+")") + default: + rr.State = RestApplied + end = latest(end, rr.Reported) + } + sort.Strings(rr.Modules) + w.Rest = append(w.Rest, *rr) + } + sort.SliceStable(w.Rest, func(a, b int) bool { + if w.Rest[a].Node != w.Rest[b].Node { + return w.Rest[a].Node < w.Rest[b].Node + } + return w.Rest[a].Sent.Before(w.Rest[b].Sent) + }) switch { case len(failed) > 0: return append(out, point{phase: PhaseApply, tier: -1, said: "not applied on " + strings.Join(failed, ", ")}) @@ -393,7 +470,8 @@ func (p Plan) points(now time.Time, applied AppliedLookup, w *WalkPhases) []poin type restSend struct { module, node string - at time.Time + walkSent time.Time + what SentDeclaration } func askedAnyOf(p Plan, tier []string) bool { @@ -451,28 +529,79 @@ func words(d time.Duration) string { return d.Round(100 * time.Millisecond).String() } -// FirstAppliedAfter is a machine's first report of a send made at or after a walk's send to it, within -// ApplySilentAfter of it — the controller's `apply` durations, which measure every send to its first report (to-be -// 45 Phase 0). A declaration sent later carries the build too, so its report counts; one sent past the bound is a -// machine that slept, listed as silent and never moving the walk's end (ADR 0282 decision 1). -func (i *Inventory) FirstAppliedAfter(ctx context.Context, node string, sent time.Time) (AppliedReport, bool, error) { - var started time.Time - var ms int64 - var outcome string - err := i.store.Pool().QueryRow(ctx, - `select started, took_ms, detail from duration +// FirstAppliedAfter is a machine's report of the send a walk made to it: the controller's `apply` durations, which +// measure every send to its first report (to-be 45 Phase 0), each row starting at the machine's own send time and +// naming the declaration it was about. send is what the walk kept of that send; walkSent its own record of the +// send to the rest, made after every machine of the rest was sent (novox/hq issue 582). FirstReportOf chooses. +func (i *Inventory) FirstAppliedAfter(ctx context.Context, node string, send SentDeclaration, walkSent time.Time) (AppliedReport, bool, error) { + from, to := applyRowsBetween(send, walkSent) + rows, err := i.store.Pool().Query(ctx, + `select ref, started, took_ms, detail from duration where kind = $1 and subject = $2 and started >= $3 and started < $4 - order by started limit 1`, - DurationApply, node, sent.Add(-appliedSlack), sent.Add(ApplySilentAfter)).Scan(&started, &ms, &outcome) - if errors.Is(err, pgx.ErrNoRows) { - return AppliedReport{}, false, nil - } + order by started`, + DurationApply, node, from, to) if err != nil { return AppliedReport{}, false, err } - return AppliedReport{At: started.Add(time.Duration(ms) * time.Millisecond).UTC(), Outcome: outcome}, true, nil + defer rows.Close() + var ds []Duration + for rows.Next() { + d := Duration{Kind: DurationApply, Subject: node, Node: node} + var ms int64 + if err := rows.Scan(&d.Ref, &d.Started, &ms, &d.Detail); err != nil { + return AppliedReport{}, false, err + } + d.Took = time.Duration(ms) * time.Millisecond + ds = append(ds, d) + } + if err := rows.Err(); err != nil { + return AppliedReport{}, false, err + } + r, ok := FirstReportOf(ds, send, walkSent) + return r, ok, nil } -// appliedSlack is how much earlier than a walk's record of its send the machine's own record of it may be: -// the send is recorded on the machine first, then on the walk. +// applyRowsBetween is the span of a machine's `apply` durations that may hold its report of a walk's send. +func applyRowsBetween(send SentDeclaration, walkSent time.Time) (time.Time, time.Time) { + switch { + case !send.At.IsZero(): + return send.At.Add(-time.Millisecond), send.At.Add(time.Millisecond) + case send.Digest != "": + return walkSent.Add(-restSendBound), walkSent.Add(ApplySilentAfter) + } + return walkSent.Add(-appliedSlack), walkSent.Add(ApplySilentAfter) +} + +// FirstReportOf picks, from a machine's `apply` durations ordered by their start, its report of the send a walk made +// to it (novox/hq issue 582): +// - with the machine's own send time kept (every walk since issue 582): the duration of that very send, and of the +// declaration sent where the walk kept its digest. A report of a later send is never this walk's: a send made +// later, by another walk, would end this walk at that walk's send. +// - with only the declaration's digest kept (a walk kept before): the first duration about that declaration +// within restSendBound before the walk's record of its send, the machine sent first in the rest included. +// - with neither (a walk kept before ADR 0282's rest): the first duration started no earlier than appliedSlack +// before the walk's record of its send, as it was read before. +// +// A send past ApplySilentAfter is never read: its machine slept, and is listed silent (ADR 0282 decision 1). +func FirstReportOf(ds []Duration, send SentDeclaration, walkSent time.Time) (AppliedReport, bool) { + from, to := applyRowsBetween(send, walkSent) + for _, d := range ds { + if d.Started.Before(from) || !d.Started.Before(to) { + continue + } + if send.Digest != "" && !strings.HasPrefix(d.Ref, send.Digest+"@") { + continue + } + return AppliedReport{At: d.Started.Add(d.Took).UTC(), Outcome: d.Detail}, true + } + return AppliedReport{}, false +} + +// appliedSlack is how much earlier than a walk's record of its send the machine's own record of it may be, for a +// walk that kept neither the machine's send time nor what it sent. const appliedSlack = 2 * time.Second + +// restSendBound is how long before a walk's record of its send to the rest a machine of the rest may have been +// sent, for a walk that kept what it sent and not when: the send to the rest goes machine by machine, and took +// 19 to 28 seconds on 2026-10-12 (novox/hq issue 582). +const restSendBound = 5 * time.Minute diff --git a/internal/inventory/walkphases_582_test.go b/internal/inventory/walkphases_582_test.go new file mode 100644 index 00000000..a52a8e8f --- /dev/null +++ b/internal/inventory/walkphases_582_test.go @@ -0,0 +1,188 @@ +package inventory + +import ( + "strings" + "testing" + "time" +) + +// novox/hq issue 582: the walk's own record of its send to the rest is made after every machine of the rest was +// sent, and that send took 19 to 28 s on 2026-10-12. The machine sent first reported 5 s after its own send, long +// before the walk's record; read from the walk's record, its report was missed and the walk ended at another walk's +// send to it, minutes later. A walk of one tier and three machines of the rest, the send to the rest taking 20 s. +type sendRest582 struct { + judged, walkSent time.Time + walk Plan + rows map[string][]Duration +} + +func walk582(t *testing.T) sendRest582 { + t.Helper() + judged := time.Date(2026, 10, 12, 17, 45, 35, 137000000, time.UTC) + ownSend := map[string]time.Duration{"g14": 500 * time.Millisecond, "novox": 10 * time.Second, "shanks": 20 * time.Second} + walkSent := judged.Add(20200 * time.Millisecond) + merged := judged.Add(-10 * time.Minute) + built, first := judged.Add(-3*time.Minute), judged.Add(-2*time.Minute) + rest := map[string]SentDeclaration{} + rows := map[string][]Duration{} + took := map[string]time.Duration{"g14": 5 * time.Second, "novox": 13 * time.Second, "shanks": 15 * time.Second} + for i, n := range []string{"g14", "novox", "shanks"} { + at := judged.Add(ownSend[n]) + digest := "walk-" + n + rest[n] = SentDeclaration{Digest: digest, Sequence: int64(647 + i), At: at} + rows[n] = append(rows[n], + // This walk's send, and its report. + Duration{Kind: DurationApply, Subject: n, Node: n, Ref: digest + "@" + at.Format("2006-01-02T15:04:05.000000"), + Started: at, Took: took[n], Detail: OutcomeApplied}, + // The next walk's send to the same machine, nine minutes later, and its report. + Duration{Kind: DurationApply, Subject: n, Node: n, Ref: "next-" + n + "@x", + Started: judged.Add(9*time.Minute + ownSend[n]), Took: 4 * time.Second, Detail: OutcomeApplied}) + } + p := Plan{ID: "plan-582", State: PlanDone, Tier: 1, Tiers: [][]string{{"build-agent"}}, Created: merged, + Delivery: &PlanDelivery{Merges: []PlanMerge{{Repository: "novox/mesh-controller", Commit: "c", Merged: merged}}}, + Times: &PlanTimes{Cut: &merged, Class: ClassCore}, + Modules: map[string]*PlanModule{"build-agent": {State: "built", AskedAt: &merged, BuiltAt: &built, + FirstAt: &first, SentAt: &walkSent, Gate: &PlanGate{JudgedAt: &judged}, Rest: rest}}} + return sendRest582{judged: judged, walkSent: walkSent, walk: p, rows: rows} +} + +// lookup answers from the rows as the controller's durations would. +func (w sendRest582) lookup(node string, send SentDeclaration, walkSent time.Time) (AppliedReport, bool) { + return FirstReportOf(w.rows[node], send, walkSent) +} + +// The apply phase ends at the last machine's own report of this walk's send — shanks, sent at the end of the 20 s +// and reporting 15 s later — and never at the next walk's send; every machine's report is shown. +func TestTheApplyPhaseEndsAtEachMachinesOwnReport(t *testing.T) { + c := walk582(t) + w := c.walk.WalkPhases(c.judged.Add(time.Hour), c.lookup) + sumsUp(t, w) + want := c.judged.Add(35 * time.Second) // shanks: sent 20 s after the judgement, reported 15 s later + if !w.End.Equal(want) { + t.Fatalf("the walk ends at %s, want %s: its rest %+v", w.End, want, w.Rest) + } + if got := phase(w, PhaseApply, -1); got.State != PhaseMeasured || got.TookMS != want.Sub(c.walkSent).Milliseconds() { + t.Fatalf("apply: %+v", got) + } + if len(w.Rest) != 3 || len(w.Silent) != 0 { + t.Fatalf("every machine of the rest is shown with its report: %+v, silent %v", w.Rest, w.Silent) + } + g14 := w.Rest[0] + if g14.Node != "g14" || g14.State != RestApplied || g14.Reported == nil || + !g14.Reported.Equal(c.judged.Add(5500*time.Millisecond)) || g14.Took != "5s" || g14.Sequence != 647 || + !g14.Sent.Equal(c.judged.Add(500*time.Millisecond)) || g14.SentSaid != "" { + t.Fatalf("g14, sent first and reporting 5 s after its own send: %+v", g14) + } +} + +// A walk kept before its machines' own send times were: matched by the declaration it sent, within restSendBound +// of its record of the send, so the machine sent first is still found; a walk that kept neither is read as before. +func TestAnOlderWalkIsReadByWhatItSent(t *testing.T) { + c := walk582(t) + for n, s := range c.walk.Modules["build-agent"].Rest { + s.At = time.Time{} + c.walk.Modules["build-agent"].Rest[n] = s + } + w := c.walk.WalkPhases(c.judged.Add(time.Hour), c.lookup) + sumsUp(t, w) + if !w.End.Equal(c.judged.Add(35 * time.Second)) { + t.Fatalf("by the declaration sent: end %s, rest %+v", w.End, w.Rest) + } + if !strings.Contains(w.Rest[0].SentSaid, "walk's record") || !w.Rest[0].Sent.Equal(c.walkSent) { + t.Fatalf("a send time not kept is said: %+v", w.Rest[0]) + } + // Nothing of the send kept: the first report no earlier than appliedSlack before the walk's record, as before. + if r, ok := FirstReportOf(c.rows["shanks"], SentDeclaration{}, c.walkSent); !ok || !r.At.Equal(c.judged.Add(35*time.Second)) { + t.Fatalf("a walk that kept nothing of its send: %+v %v", r, ok) + } +} + +// A machine that said nothing of this walk's send is waiting, then silent past ApplySilentAfter: the report of a +// later send to it is never taken as this walk's, and never moves the walk's end. +func TestALaterSendsReportIsNeverThisWalks(t *testing.T) { + c := walk582(t) + c.rows["g14"] = c.rows["g14"][1:] // g14 slept through this walk's send; the next walk's send it reported + w := c.walk.WalkPhases(c.judged.Add(5*time.Minute), c.lookup) + if w.End != nil || !strings.Contains(phase(w, PhaseApply, -1).Said, "waiting for g14") { + t.Fatalf("g14 not heard from yet: end %v, apply %+v", w.End, phase(w, PhaseApply, -1)) + } + w = c.walk.WalkPhases(c.judged.Add(time.Hour), c.lookup) + sumsUp(t, w) + if len(w.Silent) != 1 || w.Silent[0] != "g14" || !w.End.Equal(c.judged.Add(35*time.Second)) { + t.Fatalf("g14 silent, the end at shanks' report: silent %v, end %s", w.Silent, w.End) + } + if w.Rest[0].State != RestSilent || w.Rest[0].Reported != nil { + t.Fatalf("g14's later report taken as this walk's: %+v", w.Rest[0]) + } +} + +// A machine sent by two tiers of one walk that said nothing of the first send has said it of the second, which +// carries the first tier's build too. +func TestAMachineSentTwiceByOneWalkIsHeardByItsLaterSend(t *testing.T) { + c := walk582(t) + second := c.judged.Add(4 * time.Minute) + secondSent := second.Add(time.Second) + c.walk.Tiers = append(c.walk.Tiers, []string{"route-proxy"}) + c.walk.Tier = 2 + asked, built := c.judged.Add(time.Minute), c.judged.Add(2*time.Minute) + first := built.Add(time.Second) + c.walk.Modules["route-proxy"] = &PlanModule{State: "built", AskedAt: &asked, BuiltAt: &built, FirstAt: &first, + SentAt: &secondSent, Gate: &PlanGate{JudgedAt: &second}, + Rest: map[string]SentDeclaration{"novox": {Digest: "walk-novox-2", Sequence: 1554, At: second.Add(500 * time.Millisecond)}}} + c.rows["novox"] = []Duration{{Kind: DurationApply, Subject: "novox", Ref: "walk-novox-2@x", + Started: second.Add(500 * time.Millisecond), Took: 14 * time.Second, Detail: OutcomeApplied}} + w := c.walk.WalkPhases(c.judged.Add(time.Hour), c.lookup) + sumsUp(t, w) + want := second.Add(14500 * time.Millisecond) + if !w.End.Equal(want) || len(w.Silent) != 0 { + t.Fatalf("end %s (want %s), silent %v, rest %+v", w.End, want, w.Silent, w.Rest) + } + for _, r := range w.Rest { + if r.Node == "novox" && (r.State != RestApplied || !r.Reported.Equal(want)) { + t.Fatalf("novox, heard by its later send of this walk: %+v", r) + } + } +} + +// The store: a machine's own send time is read with what it was sent, and its `apply` duration is found by it — +// the send and the duration start at one moment, the node's sent_at. +func TestAWalksRestIsReadFromEachMachinesOwnSend(t *testing.T) { + inv := ForTest(t) + ctx := t.Context() + record, err := inv.AddNode(ctx, "g14") + if err != nil { + t.Fatal(err) + } + if _, err := inv.NextSequence(ctx, record.ID); err != nil { + t.Fatal(err) + } + if err := inv.RecordSent(ctx, record.ID, "d-582", map[string]string{"build-agent": "c1"}); err != nil { + t.Fatal(err) + } + sent, found, err := inv.SentTo(ctx, "g14") + if err != nil || !found || sent.At.IsZero() || sent.Digest != "d-582" { + t.Fatalf("the machine's own send time: %+v %v %v", sent, found, err) + } + if _, ok, err := inv.FirstAppliedAfter(ctx, "g14", sent, sent.At.Add(20*time.Second)); err != nil || ok { + t.Fatalf("no report yet: %v %v", ok, err) + } + if err := inv.RecordApplyDuration(ctx, "g14", "d-582", OutcomeApplied); err != nil { + t.Fatal(err) + } + // The walk's record of its send, 20 s after the machine's own: still found, by the machine's own send. + r, ok, err := inv.FirstAppliedAfter(ctx, "g14", sent, sent.At.Add(20*time.Second)) + if err != nil || !ok || r.Outcome != OutcomeApplied || r.At.Before(sent.At) { + t.Fatalf("the machine's own report of its send: %+v %v %v", r, ok, err) + } + // A later send, reported: never this walk's. + if err := inv.RecordSent(ctx, record.ID, "d-next", map[string]string{"build-agent": "c2"}); err != nil { + t.Fatal(err) + } + if err := inv.RecordApplyDuration(ctx, "g14", "d-next", OutcomeApplied); err != nil { + t.Fatal(err) + } + again, _, _ := inv.FirstAppliedAfter(ctx, "g14", sent, sent.At.Add(20*time.Second)) + if !again.At.Equal(r.At) { + t.Fatalf("a later send's report read as this walk's: %+v, was %+v", again, r) + } +} diff --git a/internal/inventory/walkphases_test.go b/internal/inventory/walkphases_test.go index ca045749..afef0148 100644 --- a/internal/inventory/walkphases_test.go +++ b/internal/inventory/walkphases_test.go @@ -116,7 +116,7 @@ func TestAWalkOfTwoTiersEndsAtItsRestsLastReport(t *testing.T) { }} reports := map[string]AppliedReport{"ace": {At: *at("05:12"), Outcome: OutcomeApplied}, "shanks": {At: *at("08:15.25"), Outcome: OutcomeApplied}} - w := p.WalkPhases(*at("30:00"), func(node string, _ time.Time) (AppliedReport, bool) { + w := p.WalkPhases(*at("30:00"), func(node string, _ SentDeclaration, _ time.Time) (AppliedReport, bool) { r, ok := reports[node] return r, ok }) @@ -135,7 +135,7 @@ func TestAWalkOfTwoTiersEndsAtItsRestsLastReport(t *testing.T) { } // A report not yet in: the walk has no end, and its apply is unknown — never zero. delete(reports, "shanks") - w = p.WalkPhases(*at("10:00"), func(node string, _ time.Time) (AppliedReport, bool) { + w = p.WalkPhases(*at("10:00"), func(node string, _ SentDeclaration, _ time.Time) (AppliedReport, bool) { r, ok := reports[node] return r, ok }) @@ -146,7 +146,7 @@ func TestAWalkOfTwoTiersEndsAtItsRestsLastReport(t *testing.T) { t.Errorf("apply while shanks has not reported: %+v", got) } // Silent past the bound: listed, not waited for. - w = p.WalkPhases(at("08:00.5").Add(ApplySilentAfter+time.Second), func(node string, _ time.Time) (AppliedReport, bool) { + w = p.WalkPhases(at("08:00.5").Add(ApplySilentAfter+time.Second), func(node string, _ SentDeclaration, _ time.Time) (AppliedReport, bool) { r, ok := reports[node] return r, ok }) @@ -156,7 +156,7 @@ func TestAWalkOfTwoTiersEndsAtItsRestsLastReport(t *testing.T) { } // A failed report: the walk has no end, said. reports["shanks"] = AppliedReport{At: *at("08:10"), Outcome: OutcomeFailed} - w = p.WalkPhases(*at("30:00"), func(node string, _ time.Time) (AppliedReport, bool) { + w = p.WalkPhases(*at("30:00"), func(node string, _ SentDeclaration, _ time.Time) (AppliedReport, bool) { r, ok := reports[node] return r, ok }) @@ -247,7 +247,7 @@ func TestAWalksMomentsAreKeptAndItsRestsReportRead(t *testing.T) { t.Fatal("a merge's time was not kept") } sent := time.Now().UTC().Add(-time.Minute).Truncate(time.Millisecond) - if _, ok, err := inv.FirstAppliedAfter(ctx, "ace", sent); err != nil || ok { + if _, ok, err := inv.FirstAppliedAfter(ctx, "ace", SentDeclaration{}, sent); err != nil || ok { t.Fatalf("no report yet: %v %v", ok, err) } for _, d := range []Duration{ @@ -258,7 +258,7 @@ func TestAWalksMomentsAreKeptAndItsRestsReportRead(t *testing.T) { t.Fatal(err) } } - r, ok, err := inv.FirstAppliedAfter(ctx, "ace", sent) + r, ok, err := inv.FirstAppliedAfter(ctx, "ace", SentDeclaration{}, sent) if err != nil || !ok || r.Outcome != OutcomeApplied || !r.At.Equal(sent.Add(11*time.Second)) { t.Fatalf("the report after the send: %+v %v %v", r, ok, err) } @@ -281,7 +281,7 @@ func TestALateReportIsSilentNotTheEnd(t *testing.T) { p := walkOf(t, recordedWalk) p.Modules["mesh-delivery"].SentAt = &sent p.Modules["mesh-delivery"].Rest = map[string]SentDeclaration{"laptop": {Digest: "d"}, "server": {Digest: "e"}} - w := p.WalkPhases(sent.Add(3*time.Hour), func(node string, _ time.Time) (AppliedReport, bool) { + w := p.WalkPhases(sent.Add(3*time.Hour), func(node string, _ SentDeclaration, _ time.Time) (AppliedReport, bool) { if node == "server" { return AppliedReport{At: sent.Add(20 * time.Second), Outcome: OutcomeApplied}, true } @@ -296,7 +296,7 @@ func TestALateReportIsSilentNotTheEnd(t *testing.T) { Started: sent.Add(time.Hour), Took: time.Second, Detail: OutcomeApplied}); err != nil { t.Fatal(err) } - if _, ok, err := inv.FirstAppliedAfter(ctx, "laptop", sent); err != nil || ok { + if _, ok, err := inv.FirstAppliedAfter(ctx, "laptop", SentDeclaration{}, sent); err != nil || ok { t.Fatalf("a send past the bound read as the walk's: %v %v", ok, err) } } diff --git a/testdata/store-tests b/testdata/store-tests index b9f34a82..14d5ada0 100644 --- a/testdata/store-tests +++ b/testdata/store-tests @@ -36,3 +36,4 @@ ./cmd/mesh-controller TestAGateJudgesTheOtherMovesBesideAKeptCore ./cmd/mesh-controller TestReplay401 ./cmd/mesh-controller TestReplay201ThePlansSend +./internal/inventory TestAWalksRestIsReadFromEachMachinesOwnSend