DLV-F1r, DLV-F3r (indexer): the indexer reads the placement through the assigner's first view
The indexer no longer reads the assigner's placement bucket (ADR 0615 §1): it asks mesh-delivery-assigner.placement at its start, waiting at most the two-minute bound so a normal start computes the fingerprints once, then follows assigner-views on its durable consumer, asking once more on a gap. Past the bound it raises its finding naming the assigner and placement, cleared by the answer. Its status is kept in its state first-views, read by indexer_first_views; indexer_first_view_again asks again by hand. sdk/views gains the event's write time, the status hook, and the status state and tools a reader's bundle serves. Issue 1966. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
@@ -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.
|
||||
|
||||
@@ -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() })
|
||||
}
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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",
|
||||
|
||||
+1
-1
@@ -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
|
||||
|
||||
+22
-5
@@ -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, ""
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}},
|
||||
}
|
||||
}
|
||||
@@ -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"`
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user