diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 19e15c0..0c7341d 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -214,7 +214,9 @@ func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(a return } if e == nil { - settled <- struct{}{} + if !optional { // the puts never gate the first pass, so never the wants (ADR 0546) + settled <- struct{}{} + } continue } var value []byte @@ -228,13 +230,12 @@ func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(a }() return nil } - followed := watched + // The puts' watch is not waited for before the first pass: a watch of them that ends before it handed what stands only + // stops the puts, never the wants. if kvs[delivery.AssignerPutsBucket] != nil { if err := watch(delivery.AssignerPutsBucket); err != nil { logger.Printf("%v: no put is taken further until it is opened again", err) delete(kvs, delivery.AssignerPutsBucket) - } else { - followed = append(append([]string{}, watched...), delivery.AssignerPutsBucket) } } // The puts' bucket is made by the controller from the manifest's states at its start (ADR 0201), and the assigner's @@ -274,7 +275,7 @@ func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(a if err := watch(delivery.ManifestsBucket, jetstream.MetaOnly()); err != nil { return err } - for range len(followed) + 1 { + for range len(watched) + 1 { select { case <-ctx.Done(): return nil diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go index 4f6092c..3c346ab 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go @@ -5,15 +5,18 @@ import ( "errors" "fmt" "io" + "log" "os" "path/filepath" "reflect" "slices" "strings" "testing" + "time" "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-sdk/go/asks" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -74,6 +77,32 @@ func TestOpensWhatItsManifestNames(t *testing.T) { t.Errorf("%s refused with %v: the start answered %v", c.missing, c.err, err) } } + // And a watch of the puts that ends before it handed what stands (its subscription closed, its grant withdrawn) does not + // keep the first pass from being made: the wants are written all the same. + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + wrote := make(chan string, 8) + kvs := buckets{} + for _, name := range append(append([]string{}, readBuckets...), ownBuckets...) { + kvs[name] = &watchKV{memKV: &memKV{values: map[string][]byte{}}} + } + ckey, _ := delivery.ComposedNodeKey("ace") + cd, err := delivery.EncodeComposedDigest(delivery.ComposedDigest{Digest: "sha256:" + strings.Repeat("d", 64), Judgement: delivery.JudgedUnchanged}) + if err != nil { + t.Fatal(err) + } + kvs[delivery.ControllerComposedBucket].(*watchKV).entries = []kvEntry{{key: ckey, value: cd}} + kvs[delivery.AssignerPutsBucket].(*watchKV).closed = true + kvs[delivery.AssignerWantsBucket].(*watchKV).wrote = wrote + go follow(ctx, nil, kvs, func(asks.Condition) error { return nil }, nil, log.New(io.Discard, "", 0)) + select { + case key := <-wrote: + if key != "want.ace.node" { + t.Errorf("the first pass wrote %s", key) + } + case <-ctx.Done(): + t.Error("a puts watch that ended before its first marker kept the first pass from being made: no want was written") + } } // 1297/B2a: without a credential the program does not start, and says why. @@ -97,3 +126,47 @@ func (r refusingJS) KeyValue(_ context.Context, name string) (jetstream.KeyValue } return &memKV{values: map[string][]byte{}}, nil } + +// watchKV is a memKV whose watch hands its entries and then the marker that all stands was handed, or ends at once when +// closed; a put on it is said on wrote. +type watchKV struct { + *memKV + entries []kvEntry + closed bool + wrote chan string +} + +func (w *watchKV) WatchAll(context.Context, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { + ch := make(chan jetstream.KeyValueEntry, len(w.entries)+1) + if w.closed { + close(ch) + return keyWatcher{ch}, nil + } + for _, e := range w.entries { + ch <- e + } + ch <- nil + return keyWatcher{ch}, nil +} + +func (w *watchKV) Put(ctx context.Context, key string, value []byte) (uint64, error) { + if w.wrote != nil { + w.wrote <- key + } + return 1, nil +} + +type keyWatcher struct{ ch chan jetstream.KeyValueEntry } + +func (k keyWatcher) Updates() <-chan jetstream.KeyValueEntry { return k.ch } +func (k keyWatcher) Stop() error { return nil } + +type kvEntry struct { + jetstream.KeyValueEntry + key string + value []byte +} + +func (e kvEntry) Key() string { return e.key } +func (e kvEntry) Value() []byte { return e.value } +func (e kvEntry) Operation() jetstream.KeyValueOp { return jetstream.KeyValuePut }