diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/views.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/views.go index cf164ff7..33717966 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/views.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/views.go @@ -24,7 +24,8 @@ func viewTools(getenv func(string) string, state func(string) views.Source, now for _, t := range index.ReadTools { out = append(out, views.ReadTool(delivery.IndexerModule, t.Tool, t.State, t.Words, state(t.State), marks, now)) } - return out + // Its own first views, of the assigner's placement (DLV-F3r): read, and asked again by hand. + return append(out, views.StatusTools(delivery.IndexerModule, "indexer_first_views", "indexer_first_view_again", stdio.State(views.StatusState), now)...) } // keptSource is one of the indexer's states as its runtime reads it. diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go index 1234d824..3a228678 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go @@ -287,6 +287,8 @@ func startIndexer(t *testing.T, put func(context.Context, buildLog), before ...f b(ctx, js) } + serveAssignerPlacement(t, ctx, js, ctl) + broker := filepath.Join(t.TempDir(), "broker") cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-indexer", User: "n1.mesh-delivery-indexer", Password: "pw"}) if err := os.WriteFile(broker, cred, 0o600); err != nil { @@ -806,3 +808,44 @@ func TestIndexerCountsOnlyThePlacedBuildNodesToolchains(t *testing.T) { t.Fatalf("within 20 s of n2's placement taken away avahi has fp2 %q, none2 %q", m.FP2, m.None2) } } + +// serveAssignerPlacement is the assigner as the owner of its placement (ADR 0615, ADR 0621), with the real owner's side: +// its first-view tool placement answered from the bucket at its mark (the bucket made empty where a test made none), +// and its views family said by the views emitter onto ASSIGNER_VIEWS, which the indexer's durable consumer reads. +func serveAssignerPlacement(t *testing.T, ctx context.Context, js jetstream.JetStream, nc *nats.Conn) { + t.Helper() + placement, err := js.KeyValue(ctx, delivery.PlacementBucket) + if err != nil { + if placement, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.PlacementBucket, History: delivery.PlacementHistory}); err != nil { + t.Fatal(err) + } + } + marks, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.AssignerModule + "_" + views.MarksState, History: 1}) + if err != nil { + t.Fatal(err) + } + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "ASSIGNER_VIEWS", Subjects: []string{"mesh.stream.mesh-delivery-assigner.event.views.>"}}); err != nil { + t.Fatal(err) + } + if _, err := js.CreateOrUpdateConsumer(ctx, "ASSIGNER_VIEWS", jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-indexer", AckPolicy: jetstream.AckExplicitPolicy}); err != nil { + t.Fatal(err) + } + em := views.Emitter{Owner: delivery.AssignerModule, Subject: func(e string) string { return process.ModuleStreamSubject(delivery.AssignerModule, e) }, + Publish: views.JetStreamPublish(js), Marks: marks} + if err := em.Start(ctx, delivery.PlacementState, placement, nil); err != nil { + t.Fatal(err) + } + sub, err := nc.Subscribe("mesh.mod.mesh-delivery-assigner.call.placement.>", func(m *nats.Msg) { + a, err := views.FirstView(delivery.AssignerModule, delivery.PlacementState, jsSource{ctx, placement}, jsSource{ctx, marks}, "", time.Now()) + out := map[string]any{"result": a, "node": "n0"} + if err != nil { + out = map[string]any{"error": err.Error()} + } + raw, _ := json.Marshal(out) + _ = m.Respond(raw) + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = sub.Unsubscribe() }) +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index a8c1f92d..4e64eb90 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -378,14 +378,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // Only the toolchain reports of nodes the catalogue places the build module on count (issue 1940, ADR 0549 §3): the // placement is settled first, so the toolchains settle through the gate already knowing it. gate := buildrun.NewToolchainGate(sortedModules(), ix.PutToolchain, logger.Printf) - placementErrs, err := buildrun.FollowPlacement(ctx, js, gate, func(ctx context.Context) { + findings := process.AwaitBucket(ctx, js, conn, findingsBucket, "no finding of the indexer is written", logger.Printf) + // The placement is the assigner's: read through its first view and its views family, never its bucket (DLV-F1r). + placementErrs := make(chan error) + followPlacement(ctx, js, cred, calls.Call, gate, func(ctx context.Context) { if err := ix.Recompute(ctx); err != nil { logger.Printf("writing the modules again after the placement of a build module changed: %v", err) } - }, logger.Printf) - if err != nil { - return err - } + }, findings, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "indexer_first_views reads nothing and indexer_first_view_again is not heard", logger.Printf), logger.Printf) tools := newToolchainWatch(ctx, js, toolchainSink{ put: gate.Toolchain, apply: func(ctx context.Context, module, key string, value []byte, deleted bool) error { @@ -496,7 +496,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // Each change of the states its readers read is said on its views family (ADR 0615 §5), from the mark each state was // said up to; a first view (the tools bundle) answers at that mark. startViews(ctx, js, cred, process.AwaitBucket(ctx, js, conn, index.MarksBucket, "no change is said on the views family, so every reader waits on its first view", logger.Printf), - process.AwaitBucket(ctx, js, conn, findingsBucket, "no finding of the indexer is written", logger.Printf), + findings, map[string]readyOf{"manifests": boundNow(manifests), "modules": boundNow(modules), "candidates": boundNow(candidates), index.VerdictsState: awaited(verdicts), delivery.IndexerPredefinedState: awaited(predefined), delivery.IndexerConfigCandidatesState: awaited(configCandidates), delivery.PackagesState: awaited(packages), diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go index 7821d743..b642d7ec 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go @@ -2,10 +2,13 @@ package main import ( "context" + "encoding/json" "time" "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" + "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/health" "git.novox.be/novox/mesh-sdk/go/process" @@ -71,3 +74,76 @@ func sortedStates(states map[string]readyOf) []string { // graphState is the module graph's state, named apart from run's bucket of the same name. const graphState = graph.State + +// firstViewsBucket is the indexer's own state of its first views (views.StatusState): what indexer_first_views reads and +// where indexer_first_view_again asks. +var firstViewsBucket = delivery.IndexerModule + "_" + views.StatusState + +// placementTool is the assigner's first-view tool of its placement, as the indexer's invokes names it (ADR 0621 §2). +const placementTool = delivery.AssignerModule + "." + delivery.PlacementState + +// followPlacement reads the seat mesh-placement's placement through the assigner's first view and then its views +// family, never its bucket (ADR 0615 §1, DLV-F1r), into gate: each key as it changes, settled once answered, with +// changed told whenever what the gate passes on changed. It waits for the first answer at most views.Bound, so a +// normal start, the two owners coming up together, computes the fingerprints once; past it the indexer goes on, the gate +// passing every report, and raises its condition naming the assigner and placement until the answer clears it. +func followPlacement(ctx context.Context, js jetstream.JetStream, cred process.Account, call func(string, any) (json.RawMessage, error), + gate *buildrun.ToolchainGate, changed func(context.Context), findings, status *process.AwaitedBucket, logf func(string, ...any)) *views.Follower { + finding := func(s views.Status, raised bool) { + kv := findings.KV() + if kv == nil { + return + } + key, value, err := views.UnansweredFinding(delivery.IndexerModule, s, time.Now()) + if err == nil && raised { + _, err = kv.Put(ctx, key, value) + } else if err == nil { + err = kv.Delete(ctx, key) + } + if err != nil { + logf("views: the finding of the first view %s is not written: %v", s.Tool, err) + } + } + f := &views.Follower{Owner: delivery.AssignerModule, Tool: placementTool, State: delivery.PlacementState, + Ask: views.AskVia(call, placementTool), Logf: logf, + Raise: func(s views.Status) { finding(s, true) }, Clear: func(s views.Status) { finding(s, false) }, + Changed: func(key string, value json.RawMessage, deleted bool, at time.Time) { + if gate.Placement(key, value, deleted, at) { + changed(ctx) + } + }} + status.Ready(func(kv jetstream.KeyValue) { + f.Said = views.KeepStatus(ctx, kv, logf) + if err := views.HearAgain(ctx, kv, []*views.Follower{f}, logf); err != nil { + logf("views: indexer_first_view_again is not heard (%v) until the indexer starts again", err) + } + }) + answered := make(chan struct{}) + go func() { + if err := f.Start(ctx); err != nil { + return + } + if gate.Settled() { + changed(ctx) + } + close(answered) + }() + // The assigner's views family, on the indexer's durable consumer, once the first view is asked. + switch cons, err := cred.EventConsumer(ctx, js, catalogue.StreamBusName(views.AssignerStream)); { + case err != nil: + logf("the consumer %s on %s does not exist yet (%v): the placement is the first view's alone, asked again by hand, until the indexer starts again", + cred.EventConsumerName(), catalogue.StreamBusName(views.AssignerStream), err) + default: + if _, err := process.HearEvents(cons, views.Hear(ctx, map[string]*views.Follower{delivery.PlacementState: f}, logf), logf); err != nil { + logf("the assigner's views family is not heard (%v): the placement is the first view's alone until the indexer starts again", err) + } + } + wait, cancel := context.WithTimeout(ctx, views.Bound) // a context's bound, no timer (ADR 0437 §8) + defer cancel() + select { + case <-answered: + case <-wait.Done(): + logf("the assigner has not answered placement within %s: the indexer goes on, every node's toolchain report counting until it answers", views.Bound) + } + return f +} diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index dba12132..5d82fa35 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -113,15 +113,16 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build } // 1126 note 76: heard keeps the last 64 FORGE messages the indexer handled, for indexer_heard. if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "predefined", History: 5}, {Name: "config-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}, {Name: "verdicts", History: 5}, {Name: "oldsections", History: 1}, {Name: delivery.PackagesState, History: delivery.PackagesHistory}, - {Name: graph.State, History: graph.History}, {Name: views.MarksState, History: 1}, {Name: health.FindingsState, History: 5}}; !reflect.DeepEqual(m.State, want) { + {Name: graph.State, History: graph.History}, {Name: views.MarksState, History: 1}, {Name: health.FindingsState, History: 5}, {Name: views.StatusState, History: 1}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } - if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-delivery-assigner.placement", "mesh-delivery-builder.build-log", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { + // DLV-F1r (ADR 0615 §1): the assigner's placement is read through its first view and views family, never its state. + if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-delivery-builder.build-log", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { t.Errorf("reads = %v, want %v", m.Reads, want) } // "Reads FORGE" is a consumes of the indexer's stream: the broker grants only buckets by reads (mesh-control // internal/broker/state.go:62-93). - if want := []string{"mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-delivery-checker.preview.written", "mesh-delivery-forge-bridge.forge.**"}; !reflect.DeepEqual(sorted(m.Consumes), want) { + if want := []string{"mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-delivery-assigner.views.placement", "mesh-delivery-checker.preview.written", "mesh-delivery-forge-bridge.forge.**"}; !reflect.DeepEqual(sorted(m.Consumes), want) { t.Errorf("consumes = %v, want %v", m.Consumes, want) } if b, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 || b != "${dir:state}/broker" { @@ -133,7 +134,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build // Three invokes: the file of a registered module's manifest and lock files (registered.go), the forge's trees (the named paths' objects, since the indexer is the one reader of a manifest's paths: // ADR 0499 clauses 2 and 3, ADR 0495 clause 1; and the module directories of the trunk, issue 1900) and whether a // commit is in another (issue 1900). - if want := []string{"gitea.gitea_commit_trees", "gitea.gitea_contains", "gitea.gitea_get_file", "gitea.gitea_open_pulls", "seat:node-engine.composition"}; !reflect.DeepEqual(m.Invokes, want) { + if want := []string{"mesh-delivery-assigner.placement", "gitea.gitea_commit_trees", "gitea.gitea_contains", "gitea.gitea_get_file", "gitea.gitea_open_pulls", "seat:node-engine.composition"}; !reflect.DeepEqual(m.Invokes, want) { t.Errorf("invokes = %v, want %v", m.Invokes, want) } // 1945: indexer_take_in, served by its tools bundle through the takeins state; the process serves none. The same bundle @@ -148,7 +149,8 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build t.Errorf("%s answers the state %s, which the views family does not say", v.Tool, v.State) } } - if !reflect.DeepEqual(m.Tools, want) || len(want) != 16 { + want = append(want, "indexer_first_views", "indexer_first_view_again") // DLV-F3r + if !reflect.DeepEqual(m.Tools, want) || len(want) != 18 { t.Errorf("tools = %v, want %v", m.Tools, want) } if m.RunsAs != m.Module || m.SecretsOwner != m.RunsAs { diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index d6a3f127..2b42a1e9 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -31,11 +31,14 @@ "indexer_read_packages", "indexer_read_graph", "indexer_read_heard", - "indexer_read_taken" + "indexer_read_taken", + "indexer_first_views", + "indexer_first_view_again" ], "consumes": [ "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", + "mesh-delivery-assigner.views.placement", "mesh-delivery-checker.preview.written", "mesh-delivery-forge-bridge.forge.**" ], @@ -53,7 +56,8 @@ { "name": "packages", "history": 5 }, { "name": "graph", "history": 5 }, { "name": "view-marks", "history": 1 }, - { "name": "findings", "history": 5 } + { "name": "findings", "history": 5 }, + { "name": "first-views", "history": 1 } ], "streams": [ { "name": "indexer-views", "events": ["views.**"], "max-age-days": 7, "max-mib": 256 } @@ -65,12 +69,12 @@ "reads": [ "mesh-build-go.toolchains", "mesh-build-node.toolchains", - "mesh-delivery-assigner.placement", "mesh-delivery-builder.build-log", "mesh-delivery-checker.previews", "mesh-delivery-gate.source" ], "invokes": [ + "mesh-delivery-assigner.placement", "gitea.gitea_commit_trees", "gitea.gitea_contains", "gitea.gitea_get_file", diff --git a/sdk/views/emit.go b/sdk/views/emit.go index 3552dbef..23a251ae 100644 --- a/sdk/views/emit.go +++ b/sdk/views/emit.go @@ -126,7 +126,7 @@ func (e Emitter) Start(ctx context.Context, state string, kv jetstream.KeyValue, // Event is the event saying u, a change of state made at made. func (e Emitter) Event(state string, made time.Time, u jetstream.KeyValueEntry) Event { - ev := Event{V: Version, Owner: e.Owner, State: state, Key: u.Key(), Revision: u.Revision(), Created: made} + ev := Event{V: Version, Owner: e.Owner, State: state, Key: u.Key(), Revision: u.Revision(), Created: made, At: u.Created().UTC()} if u.Operation() != jetstream.KeyValuePut { ev.Deleted = true return ev diff --git a/sdk/views/follow.go b/sdk/views/follow.go index df2a54de..ee793b80 100644 --- a/sdk/views/follow.go +++ b/sdk/views/follow.go @@ -28,8 +28,12 @@ type Follower struct { // nothing. Raise func(Status) Clear func(Status) - // Changed is told of each key the picture changes (deleted: the key is gone), after the change is held. - Changed func(key string, value json.RawMessage, deleted bool) + // Changed is told of each key the picture changes (deleted: the key is gone), after the change is held, with when it + // was written: the zero time for a key a first view brought (since before what the reader heard). + Changed func(key string, value json.RawMessage, deleted bool, at time.Time) + // Said is told of the status each time an ask starts, is answered or fails, and of a raise and a clear: where a + // reader keeps it on the bus for its read tool (StatusTools). + Said func(Status) Logf func(string, ...any) Now func() time.Time // Bound is how long it asks before it raises; zero is views.Bound. Wait is the first wait between two asks that @@ -88,6 +92,15 @@ func (f *Follower) init() { } } +func (f *Follower) say(s Status) { + if f.Said != nil { + f.mu.Lock() + s.Keys = len(f.values) + f.mu.Unlock() + f.Said(s) + } +} + // Status is the follower's status now. func (f *Follower) Status() Status { f.mu.Lock() @@ -173,6 +186,7 @@ func (f *Follower) askUntil(ctx context.Context, why string, byHand bool) error if raise && f.Raise != nil { f.Raise(s) } + f.say(s) pause, cancel := context.WithTimeout(ctx, wait) // a context's bound between two asks, no timer of its own <-pause.Done() cancel() @@ -189,7 +203,9 @@ func (f *Follower) askUntil(ctx context.Context, why string, byHand bool) error func (f *Follower) askOnce(ctx context.Context) error { f.mu.Lock() f.st.Asked, f.st.Asks = f.now(), f.st.Asks+1 + asked := f.st f.mu.Unlock() + f.say(asked) var first Answer values := map[string]json.RawMessage{} after := "" @@ -237,15 +253,16 @@ func (f *Follower) askOnce(ctx context.Context) error { if wasRaised && f.Clear != nil { f.Clear(s) } + f.say(s) if f.Changed != nil { for k, v := range values { if ov, ok := old[k]; !ok || string(ov) != string(v) { - f.Changed(k, v, false) + f.Changed(k, v, false, time.Time{}) } } for k := range old { if _, ok := values[k]; !ok { - f.Changed(k, nil, true) + f.Changed(k, nil, true, time.Time{}) } } } @@ -291,7 +308,7 @@ func (f *Follower) Apply(ev Event) (gap bool, why string) { f.st.Revision = ev.Revision f.mu.Unlock() if f.Changed != nil { - f.Changed(ev.Key, ev.Value, ev.Deleted) + f.Changed(ev.Key, ev.Value, ev.Deleted, ev.At) } return false, "" } diff --git a/sdk/views/status.go b/sdk/views/status.go new file mode 100644 index 00000000..19907d31 --- /dev/null +++ b/sdk/views/status.go @@ -0,0 +1,145 @@ +package views + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "slices" + "strings" + "time" + + "github.com/nats-io/nats.go/jetstream" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// StatusState is a reader's own state of its first views (design 25 § 12.2): at StatusKey each first view's Status, as +// its process last said it, and at AgainKey a hand ask of its re-read tool, which the process takes and deletes. A +// process serves no tool (ADR 0436), so its read and re-read tools, in its tools bundle, meet it here. +const StatusState = "first-views" + +// StatusKey is where a reader keeps its first view of owner's state. +func StatusKey(owner, state string) string { return owner + "." + state } + +// AgainKey is where a re-read tool asks a reader's process to ask owner's first view of state again. +func AgainKey(owner, state string) string { return "again." + owner + "." + state } + +// HandAsk is one ask of a re-read tool: who asked (the runtime's proof), why and when. +type HandAsk struct { + V int `json:"v"` + Owner string `json:"owner"` + State string `json:"state"` + Why string `json:"why"` + By string `json:"by"` + At time.Time `json:"at"` +} + +// KeepStatus is a Follower's Said that writes its status at StatusKey of kv (the reader's StatusState). +func KeepStatus(ctx context.Context, kv jetstream.KeyValue, logf func(string, ...any)) func(Status) { + return func(s Status) { + data, err := json.Marshal(s) + if err == nil { + _, err = kv.Put(ctx, StatusKey(s.Owner, s.State), data) + } + if err != nil && logf != nil { + logf("views: the status of the first view %s is not kept: %v", s.Tool, err) + } + } +} + +// HearAgain takes each hand ask written at an AgainKey of kv, those standing at the start included, asks that +// follower's first view again with the ask's words and who asked, and deletes the ask. It returns once the watch is set. +func HearAgain(ctx context.Context, kv jetstream.KeyValue, followers []*Follower, logf func(string, ...any)) error { + w, err := kv.Watch(ctx, "again.>") + if err != nil { + return err + } + go func() { + defer func() { _ = w.Stop() }() + for { + select { + case <-ctx.Done(): + return + case e, ok := <-w.Updates(): + if !ok { + return + } + if e == nil || e.Operation() != jetstream.KeyValuePut { + continue + } + var a HandAsk + if err := json.Unmarshal(e.Value(), &a); err != nil { + logf("views: the hand ask %s is not read: %v", e.Key(), err) + _ = kv.Delete(ctx, e.Key()) + continue + } + i := slices.IndexFunc(followers, func(f *Follower) bool { return f.Owner == a.Owner && f.State == a.State }) + if i < 0 { + logf("views: the hand ask %s names no first view this part holds", e.Key()) + _ = kv.Delete(ctx, e.Key()) + continue + } + logf("views: %s asks the first view %s again by hand: %s", a.By, followers[i].Tool, a.Why) + _ = kv.Delete(ctx, e.Key(), jetstream.LastRevision(e.Revision())) + f := followers[i] + go func() { _ = f.Again(ctx, a.Why+" ("+a.By+")") }() + } + } + }() + return nil +} + +// StatusTools are a reader's read tool of its first views (read) and its re-read tool (again), in its tools bundle, +// over its StatusState: read lists what the process last said of each; again writes a hand ask the process takes, +// naming the caller the runtime proved, and is recorded in the status's by-hand count and why. +func StatusTools(reader, read, again string, st stdio.KeptState, now func() time.Time) []stdio.Tool { + return []stdio.Tool{ + {Name: read, + Description: fmt.Sprintf("%s's first views of the indexer's and the assigner's states (novox/hq ADR 0615): per owner and state, "+ + "the tool it asks, the revision its picture held when last answered, when it asked and was answered, how many asks "+ + "(by hand among them) and whether its condition (unanswered past two minutes) is raised.", reader), + Input: map[string]any{"type": "object", "properties": map[string]any{}}, + Run: func(map[string]any) (any, error) { + keys, err := st.Keys() + if err != nil { + return nil, fmt.Errorf("the state %s cannot be listed: %w", StatusState, err) + } + out := []Status{} + for _, k := range keys { + if strings.HasPrefix(k, "again.") { + continue + } + e, err := st.Get(k) + if err != nil || e == nil { + continue + } + var s Status + if json.Unmarshal(e.Value, &s) == nil { + out = append(out, s) + } + } + return map[string]any{"first-views": out}, nil + }}, + {Name: again, + Description: fmt.Sprintf("Ask %s's first view of an owner's state again now, by hand (a hand act, its caller and why kept in "+ + "the status %s reads): for a picture a person doubts. The part asks once and follows the owner's events after the answer.", reader, read), + Input: map[string]any{"type": "object", "required": []string{"owner", "state", "why"}, "properties": map[string]any{ + "owner": map[string]any{"type": "string", "description": "the owner: mesh-delivery-indexer or mesh-delivery-assigner"}, + "state": map[string]any{"type": "string", "description": "the owner's state, as the read tool names it"}, + "why": map[string]any{"type": "string", "description": "why it is asked again, in plain words"}}}, + Run: func(args map[string]any) (any, error) { + owner, _ := args["owner"].(string) + state, _ := args["state"].(string) + why, _ := args["why"].(string) + if owner == "" || state == "" || strings.TrimSpace(why) == "" { + return nil, errors.New("again names the owner, the state and why") + } + a := HandAsk{V: Version, Owner: owner, State: state, Why: strings.TrimSpace(why), By: stdio.CallerOf(args), At: now().UTC()} + if _, err := st.Put(AgainKey(owner, state), a); err != nil { + return nil, fmt.Errorf("the hand ask is not written: %w", err) + } + return map[string]any{"asked": a, "read": read}, nil + }}, + } +} diff --git a/sdk/views/views.go b/sdk/views/views.go index d4bd069e..2be1f32f 100644 --- a/sdk/views/views.go +++ b/sdk/views/views.go @@ -74,6 +74,8 @@ type Event struct { Created time.Time `json:"created"` Deleted bool `json:"deleted,omitempty"` Value json.RawMessage `json:"value,omitempty"` + // At is when the change was written on the bus. + At time.Time `json:"at,omitzero"` // Gap says the value was too large to carry (MaxValueBytes): the reader asks its first view once. Gap bool `json:"gap,omitempty"` }