Ask build-agent for every module until a running builder says it writes references (issue 1886)
R4e's own delivery plan waited for ever in tier 1: the controller took every module but build-agent's own from a referenced builder key, and only R4e's builder and build modules, delivered by that same plan, write references. The builder now says so in its bucket of instances; until it does, a walk asks build-agent as before R4e, and an open walk waiting on a key the running builder cannot reference is judged again at the next look.
This commit is contained in:
@@ -9,6 +9,7 @@ package builderkeys
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
@@ -309,3 +310,99 @@ func firstLine(s string) string {
|
||||
line, _, _ := strings.Cut(strings.TrimSpace(s), "\n")
|
||||
return line
|
||||
}
|
||||
|
||||
// The builder's say of what its keys hold (novox/hq issue 1886): each running builder instance keeps a key, named by its
|
||||
// node, in the builder's own announced bucket of instances, saying what the keys it writes hold. An instance that writes
|
||||
// references (ADR 0498 §1) says WritesReferences there; a builder from before R4e keeps no such key, and has no bucket.
|
||||
// The bucket is the builder's and the shape is read by the controller alone, so neither is in mesh-sdk.
|
||||
const (
|
||||
// BuilderInstancesBucket is the builder's bucket of its running instances, `<owner module>_<name>`.
|
||||
BuilderInstancesBucket = delivery.BuilderModule + "_instances"
|
||||
// WritesReferences is the word an instance says when every done key it writes names where its artifacts are.
|
||||
WritesReferences = "references"
|
||||
)
|
||||
|
||||
// instanceSays is an instance's value in BuilderInstancesBucket: what the keys it writes hold.
|
||||
type instanceSays struct {
|
||||
Writes []string `json:"writes"`
|
||||
}
|
||||
|
||||
// saysReferences says whether any instance's value says it writes references. A value that does not decode says
|
||||
// nothing.
|
||||
func saysReferences(values [][]byte) bool {
|
||||
for _, v := range values {
|
||||
var says instanceSays
|
||||
if json.Unmarshal(v, &says) != nil {
|
||||
continue
|
||||
}
|
||||
for _, w := range says.Writes {
|
||||
if w == WritesReferences {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// SaysReferences says whether a running builder instance says, in its bucket of instances, that it writes references. No
|
||||
// bucket (the builder from before R4e, which never made it) and no key say no; a bucket that cannot be read is an error.
|
||||
func (r Reader) SaysReferences(ctx context.Context) (bool, error) {
|
||||
kv, err := r.bucket(ctx, BuilderInstancesBucket)
|
||||
if err != nil || kv == nil {
|
||||
return false, err
|
||||
}
|
||||
keys, err := kv.Keys(ctx)
|
||||
if errors.Is(err, jetstream.ErrNoKeysFound) {
|
||||
return false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("listing the builder's instances: %w", err)
|
||||
}
|
||||
var values [][]byte
|
||||
for _, key := range keys {
|
||||
e, err := kv.Get(ctx, key)
|
||||
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("the builder's instance %s: %w", key, err)
|
||||
}
|
||||
values = append(values, e.Value())
|
||||
}
|
||||
return saysReferences(values), nil
|
||||
}
|
||||
|
||||
// WatchSays calls wake for every change of the builder's bucket of instances from now on, until ctx ends: what makes the
|
||||
// look judge an open walk again when the builder starts or stops saying it writes references (issue 1886), with no
|
||||
// timer. It says false, and watches nothing, while the bucket is not there; the caller asks again when the builder
|
||||
// writes a key, which a builder that makes the bucket does.
|
||||
func WatchSays(ctx context.Context, js jetstream.JetStream, wake func()) (bool, error) {
|
||||
kv, err := js.KeyValue(ctx, BuilderInstancesBucket)
|
||||
if errors.Is(err, jetstream.ErrBucketNotFound) {
|
||||
return false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("the bucket %s: %w", BuilderInstancesBucket, err)
|
||||
}
|
||||
w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly())
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("watching %s: %w", BuilderInstancesBucket, err)
|
||||
}
|
||||
go func() {
|
||||
defer w.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case e, ok := <-w.Updates():
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if e != nil {
|
||||
wake() // a put and a delete both change what the builder says
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
return true, nil
|
||||
}
|
||||
|
||||
@@ -0,0 +1,54 @@
|
||||
//go:build integration
|
||||
|
||||
package builderkeys
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
)
|
||||
|
||||
// 1886/AC1 and AC4 on a real bus: no bucket of instances says no and watches nothing; an instance's key saying
|
||||
// references says yes, and its put and its delete each wake the watcher, with no timer.
|
||||
func TestTheBuildersSayIsReadAndWatchedOnTheBus(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
||||
defer cancel()
|
||||
js := aBus(t)
|
||||
r := Reader{JS: js}
|
||||
if says, err := r.SaysReferences(ctx); err != nil || says {
|
||||
t.Fatalf("no bucket: %v %v", says, err)
|
||||
}
|
||||
if on, err := WatchSays(ctx, js, func() {}); err != nil || on {
|
||||
t.Fatalf("no bucket is watched: %v %v", on, err)
|
||||
}
|
||||
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: BuilderInstancesBucket, History: 1})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
woken := make(chan struct{}, 4)
|
||||
if on, err := WatchSays(ctx, js, func() { woken <- struct{}{} }); err != nil || !on {
|
||||
t.Fatalf("the bucket is not watched: %v %v", on, err)
|
||||
}
|
||||
wake := func(what string) {
|
||||
t.Helper()
|
||||
select {
|
||||
case <-woken:
|
||||
case <-ctx.Done():
|
||||
t.Fatalf("%s woke nothing", what)
|
||||
}
|
||||
}
|
||||
put(t, kv, "shanks", []byte(`{"writes":["references"]}`))
|
||||
wake("a put")
|
||||
if says, err := r.SaysReferences(ctx); err != nil || !says {
|
||||
t.Fatalf("an instance saying references: %v %v", says, err)
|
||||
}
|
||||
if err := kv.Delete(ctx, "shanks"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
wake("a delete")
|
||||
if says, err := r.SaysReferences(ctx); err != nil || says {
|
||||
t.Fatalf("no instance left: %v %v", says, err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
package builderkeys
|
||||
|
||||
import "testing"
|
||||
|
||||
// 1886/AC1: a builder that keeps no say, or a say without the word, is a builder that does not write references; one
|
||||
// instance that says it is enough, since it builds every key done without a reference again (ADR 0498 §8).
|
||||
func TestTheBuilderSaysReferencesOnlyWhenAnInstanceSaysTheWord(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
values [][]byte
|
||||
want bool
|
||||
}{
|
||||
{"no instance", nil, false},
|
||||
{"an instance saying nothing", [][]byte{[]byte(`{"writes":[]}`)}, false},
|
||||
{"a value not read", [][]byte{[]byte(`{not json`)}, false},
|
||||
{"another word", [][]byte{[]byte(`{"writes":["digests"]}`)}, false},
|
||||
{"one instance says it", [][]byte{[]byte(`{"writes":["digests"]}`), []byte(`{"v":1,"writes":["references"]}`)}, true},
|
||||
}
|
||||
for _, c := range cases {
|
||||
if got := saysReferences(c.values); got != c.want {
|
||||
t.Errorf("%s: says references %v, want %v", c.name, got, c.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user