diff --git a/modules/mesh-grants/cmd/mesh-grants/main.go b/modules/mesh-grants/cmd/mesh-grants/main.go index 9e0c8153..6f7a4cb7 100644 --- a/modules/mesh-grants/cmd/mesh-grants/main.go +++ b/modules/mesh-grants/cmd/mesh-grants/main.go @@ -36,6 +36,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/announce" "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/views" "mesh-grants/internal/grants" @@ -156,6 +157,17 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro hearViews(m) return } + // The indexer's verdict events go to the verdicts' first view (ADR 0622 §2), which takes each at its revision. + if event, ok := strings.CutPrefix(m.Subject(), "mesh.mod."+grants.IndexerModule+".event."); ok && + (event == delivery.ManifestAcceptedEvent || event == delivery.ManifestRefusedEvent) { + if err := fv.HearVerdict(ctx, event, m.Data(), logger.Printf); err != nil { + logger.Printf("views: %s is not a verdict event, dropped: %v", m.Subject(), err) + _ = m.Term() + return + } + _ = m.Ack() + return + } handle(m.Subject(), m.Data()) _ = m.Ack() }, jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) })); err != nil { diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index 1eede1df..584e3190 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -2416,6 +2416,41 @@ func TestGrants(t *testing.T) { if _, found, _ := fv.Get(grants.ReadPlacement, "placement.c"); !found || fv.Followers[grants.ReadPlacement].Status().Asks != asksBefore { t.Error("the event did not reach the picture, or asked") } + // The verdicts follow the indexer's manifest-accepted and manifest-refused at the revision each carries (ADR 0622 §2): + // the next revision changes the picture with no ask; one already in it is passed over; one carrying no revision, or + // past the next, asks the first view once more. + verdicts := fv.Followers[grants.ReadVerdicts] + digest := strings.Repeat("ab", 32) + heard := func(event string, body any) { + t.Helper() + raw, _ := json.Marshal(body) + if err := fv.HearVerdict(ctx, event, raw, t.Logf); err != nil { + t.Fatalf("%s: %v", event, err) + } + } + verdictOf := func() delivery.ManifestVerdict { + raw, _, _ := fv.Get(grants.ReadVerdicts, digest) + v, _ := delivery.DecodeManifestVerdict(raw) + return v + } + asked := verdicts.Status().Asks + heard(delivery.ManifestAcceptedEvent, delivery.ManifestAccepted{V: 1, Module: "mod-z", Digest: digest, Judge: "j", Revision: 8}) + if v := verdictOf(); v.State != delivery.VerdictAccepted || v.Module != "mod-z" || verdicts.Status().Revision != 8 { + t.Errorf("manifest-accepted at the next revision: %+v at %d", v, verdicts.Status().Revision) + } + heard(delivery.ManifestRefusedEvent, delivery.ManifestRefused{V: 1, Module: "mod-z", Digest: digest, Judge: "j", Problems: []string{"p"}, Revision: 9}) + heard(delivery.ManifestAcceptedEvent, delivery.ManifestAccepted{V: 1, Module: "mod-z", Digest: digest, Judge: "j", Revision: 9}) + if v := verdictOf(); !v.Refused() || !slices.Equal(v.Problems, []string{"p"}) || verdicts.Status().Asks != asked { + t.Errorf("manifest-refused, then a revision already held: %+v, asks %d (was %d)", v, verdicts.Status().Asks, asked) + } + heard(delivery.ManifestAcceptedEvent, delivery.ManifestAccepted{V: 1, Module: "mod-z", Digest: digest, Judge: "j"}) + waitFor(t, func() bool { return verdicts.Status().Asks > asked }) + asked = verdicts.Status().Asks + heard(delivery.ManifestAcceptedEvent, delivery.ManifestAccepted{V: 1, Module: "mod-z", Digest: digest, Judge: "j", Revision: 12}) + waitFor(t, func() bool { return verdicts.Status().Asks > asked }) + if err := fv.HearVerdict(ctx, "manifest-judged", []byte("{}"), t.Logf); err == nil { + t.Error("an event that is no verdict was taken") + } // The tools: a state that is none of the five is refused; one is asked with its hand act. var acts []string record := func(what, why string) error { acts = append(acts, what+": "+why); return nil } diff --git a/modules/mesh-grants/internal/grants/firstviews.go b/modules/mesh-grants/internal/grants/firstviews.go index ab9b1923..4a6d017c 100644 --- a/modules/mesh-grants/internal/grants/firstviews.go +++ b/modules/mesh-grants/internal/grants/firstviews.go @@ -11,6 +11,7 @@ import ( "time" "git.novox.be/novox/mesh-sdk/go/asks" + "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/views" ) @@ -18,6 +19,8 @@ import ( // DLV-R1). The part reads the assigner's placement and wants and the indexer's manifests, modules and verdicts no longer // from their buckets, but by each owner's first-view tool, asked once at its start, then from the owner's views family // alone (each event one changed key with its revision), asking once more only on a gap in revisions, never on a timer. +// The verdicts alone are followed on the indexer's manifest-accepted and manifest-refused, which carry the revision +// (ADR 0622 §2), not on its views family (HearVerdict). // Started while an owner is down it keeps asking and never reads the owner's state; past the bound (two minutes) it // raises FirstViewUnanswered naming the owner and the tool, cleared by the answer. // @@ -204,6 +207,69 @@ func (v *Views) Again(ctx context.Context, state, why string) ([]string, error) return asked, nil } +// VerdictEvents are the indexer's two events the part follows the verdicts on after their first view (ADR 0622 §2), as its +// manifest's consumes name them. +var VerdictEvents = []string{IndexerModule + "." + delivery.ManifestAcceptedEvent, IndexerModule + "." + delivery.ManifestRefusedEvent} + +// HearVerdict takes one of the indexer's manifest-accepted or manifest-refused (event, its name) into the verdicts' first +// view, as one change of the verdicts state at the revision it carries (ADR 0622 §2): the manifest's digest its key, the +// verdict it says its value. A revision past the one after the picture's is a gap, and so is an event carrying no +// revision (an indexer that does not say it) or one saying no manifest above the picture's revision (the empty accepted +// verdict of a module leaving the trunk): the first view is asked once more on its own goroutine, said in the log. A +// revision already in the picture is passed over. Data that does not read is answered as an error, and nothing is asked. +func (v *Views) HearVerdict(ctx context.Context, event string, data []byte, logf func(string, ...any)) error { + f := v.Followers[ReadVerdicts] + if f == nil { + return nil + } + var verdict delivery.ManifestVerdict + var digest string + var revision uint64 + switch event { + case delivery.ManifestAcceptedEvent: + var a delivery.ManifestAccepted + if err := json.Unmarshal(data, &a); err != nil { + return fmt.Errorf("%s does not read: %w", event, err) + } + verdict = delivery.ManifestVerdict{V: delivery.Version, Module: a.Module, State: delivery.VerdictAccepted, Judge: a.Judge} + digest, revision = a.Digest, a.Revision + case delivery.ManifestRefusedEvent: + var r delivery.ManifestRefused + if err := json.Unmarshal(data, &r); err != nil { + return fmt.Errorf("%s does not read: %w", event, err) + } + verdict = delivery.ManifestVerdict{V: delivery.Version, Module: r.Module, State: delivery.VerdictRefused, Judge: r.Judge, Problems: r.Problems} + digest, revision = r.Digest, r.Revision + default: + return fmt.Errorf("%q is not a verdict event of the indexer", event) + } + held := f.Status() + gap := func(why string) error { + logf("views: a gap in %s (%s): its first view %s is asked once more", f.State, why, f.Tool) + go func() { _ = f.Gap(ctx, why) }() + return nil + } + if revision == 0 { + return gap(fmt.Sprintf("the indexer's %s of %s carries no revision", event, verdict.Module)) + } + key, err := delivery.VerdictsKey(digest) + if err != nil { + if revision <= held.Revision { + return nil + } + return gap(fmt.Sprintf("the indexer's %s of %s at revision %d names no manifest", event, verdict.Module, revision)) + } + value, err := json.Marshal(verdict) + if err != nil { + return err + } + if g, why := f.Apply(views.Event{V: 1, Owner: f.Owner, State: f.State, Key: key, Revision: revision, Created: held.Created, + Value: value, At: time.Now().UTC()}); g { + return gap(why) + } + return nil +} + // unansweredCondition is FirstViewUnanswered for one first view, raised or cleared. func unansweredCondition(fv FirstView, s views.Status, raised bool) asks.Condition { c := asks.Condition{Key: FirstViewUnanswered + "." + fv.Owner + "." + fv.State, Subject: asks.Scope{Scope: "module", ID: Module}} diff --git a/modules/mesh-grants/internal/grants/rootfree.go b/modules/mesh-grants/internal/grants/rootfree.go index 532af1fa..da933b8d 100644 --- a/modules/mesh-grants/internal/grants/rootfree.go +++ b/modules/mesh-grants/internal/grants/rootfree.go @@ -42,10 +42,15 @@ const EventAccountIssued = "mesh-vault." + buscompose.AccountIssuedEvent var Consumes = append([]string{EventKeeperRaised, EventKeeperCleared, EventAccountIssued, BusModule + "." + BrokerCreated, BusModule + "." + BrokerDropped, BusModule + "." + BrokerRecreated}, viewsEvents()...) -// viewsEvents are the owners' views events the part follows its first views by, `.views.`. +// viewsEvents are the owners' events the part follows its first views by: `.views.`, but the verdicts by +// the indexer's manifest-accepted and manifest-refused (VerdictEvents, ADR 0622 §2). func viewsEvents() []string { var out []string for _, fv := range FirstViews { + if fv.Read == ReadVerdicts { + out = append(out, VerdictEvents...) + continue + } out = append(out, fv.Owner+".views."+fv.State) } return out diff --git a/modules/mesh-grants/module.json b/modules/mesh-grants/module.json index 5d0f7dae..3819e9aa 100644 --- a/modules/mesh-grants/module.json +++ b/modules/mesh-grants/module.json @@ -21,7 +21,8 @@ "mesh-delivery-assigner.views.wants", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", - "mesh-delivery-indexer.views.verdicts" + "mesh-delivery-indexer.manifest-accepted", + "mesh-delivery-indexer.manifest-refused" ], "uses": [ "operator-channel"