mesh-tools
5c94759cc0
mesh/merge-gate head 5c94759cc0c2 merged into trunk c8539bf35a8d — pass: builds mesh-controller → novox; no bus step; every machine composes with the chang…
mesh/repo-check head 5c94759cc0c2 merged into trunk c8539bf35a8d — pass: its merge-check.sh passed
mesh/delivery delivered
76 lines
2.3 KiB
Go
76 lines
2.3 KiB
Go
//go:build integration
|
|
|
|
package main
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
|
|
"github.com/novox/mesh-controller/internal/chain"
|
|
"github.com/novox/mesh-controller/internal/link"
|
|
)
|
|
|
|
// withoutJetStream is a bucket whose reads fail the way a bus answers while its streams are away.
|
|
type withoutJetStream struct {
|
|
chain.Backend
|
|
away atomic.Bool
|
|
}
|
|
|
|
func (w *withoutJetStream) Get(ctx context.Context, key string) (chain.Entry, bool, error) {
|
|
if w.away.Load() {
|
|
return chain.Entry{}, false, nats.ErrJetStreamNotEnabled
|
|
}
|
|
return w.Backend.Get(ctx, key)
|
|
}
|
|
|
|
func (w *withoutJetStream) Keys(ctx context.Context) ([]string, error) {
|
|
if w.away.Load() {
|
|
return nil, nats.ErrJetStreamNotEnabled
|
|
}
|
|
return w.Backend.Keys(ctx)
|
|
}
|
|
|
|
func (w *withoutJetStream) Watch(ctx context.Context) (<-chan chain.Entry, error) {
|
|
if w.away.Load() {
|
|
return nil, nats.ErrJetStreamNotEnabled
|
|
}
|
|
return w.Backend.Watch(ctx)
|
|
}
|
|
|
|
func (w *withoutJetStream) WatchFiltered(ctx context.Context, filter string) (<-chan chain.Entry, error) {
|
|
if w.away.Load() {
|
|
return nil, nats.ErrJetStreamNotEnabled
|
|
}
|
|
return w.Backend.WatchFiltered(ctx, filter)
|
|
}
|
|
|
|
// **A merge heard while the bus answers "jetstream not enabled" is held, recorded nowhere, and recorded when handed
|
|
// over again** (issue 1583, mesh-catalog #404): hearMerge answers ErrTryAgain, so the bus does not acknowledge the
|
|
// announcement, and the merge joins a batch when the bus is back.
|
|
func TestAMergeHeardWhileTheBusHasNoJetStreamIsHeldAndRecordedWhenHandedOverAgain(t *testing.T) {
|
|
open := windowed(t)
|
|
everyHeadWaits(t)
|
|
asksWithPaths(t)
|
|
ctx := t.Context()
|
|
assignments := &withoutJetStream{Backend: chain.NewMemory()}
|
|
open.inventory.UseChain(chain.New(chain.NewMemory(), chain.NewMemory(), assignments, func(context.Context) error { return nil }))
|
|
merge := catalogueMerge("cat2bbbbbbbb", "app", t0)
|
|
|
|
assignments.away.Store(true)
|
|
err := (following{open: open}).hearMerge(ctx, merge, t0.Add(time.Second))
|
|
if !errors.Is(err, link.ErrTryAgain) {
|
|
t.Fatalf("a merge heard while the bus had no jetstream answered %v, which the bus takes as done", err)
|
|
}
|
|
|
|
assignments.away.Store(false)
|
|
hear(t, open, merge, t0.Add(10*time.Second))
|
|
if _, bs := walks(t, open); len(bs) != 1 {
|
|
t.Fatalf("the merge handed over again made %d batch(es)", len(bs))
|
|
}
|
|
}
|