jschoubben
dc024ae552
JetStreamAsks.WatchAsk opens the watch as a kept call and takes a closed watch up after the last revision heard; the first fault arms a key of the state bounds with its own time to live (StepsBoundKey), whose end while the read is unanswered raises call-unanswered about the ask, and the answer ends it. Builds: design 47 G §568, A-77 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
89 lines
3.2 KiB
Go
89 lines
3.2 KiB
Go
package buildrun
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
|
|
"git.novox.be/novox/mesh-sdk/go/busretry"
|
|
)
|
|
|
|
type fakeEntry struct {
|
|
jetstream.KeyValueEntry
|
|
rev uint64
|
|
value []byte
|
|
}
|
|
|
|
func (e fakeEntry) Revision() uint64 { return e.rev }
|
|
func (e fakeEntry) Value() []byte { return e.value }
|
|
func (e fakeEntry) Operation() jetstream.KeyValueOp { return jetstream.KeyValuePut }
|
|
|
|
type fakeWatcher struct {
|
|
jetstream.KeyWatcher
|
|
updates chan jetstream.KeyValueEntry
|
|
}
|
|
|
|
func (w fakeWatcher) Updates() <-chan jetstream.KeyValueEntry { return w.updates }
|
|
func (w fakeWatcher) Stop() error { return nil }
|
|
|
|
// Design 47 G §568: the read of an ask's steps is a kept call. A transport fault opening the watch ends no key (the wait is
|
|
// not failed): it is opened again once the bus answers, said unanswered meanwhile and answered after; and a watch that closes
|
|
// is taken up again from the revision after the last it handed over, so the step that ended meanwhile is heard.
|
|
func TestTheWatchOfAStepResumesAfterAFault(t *testing.T) {
|
|
t.Cleanup(busretry.Waits(20*time.Millisecond, time.Millisecond, 2*time.Millisecond))
|
|
passed, _ := buildstep.EncodeStep(buildstep.Step{ID: "mvl0frv18d1958", Tool: "build", Node: "ace", State: buildstep.StepPassed})
|
|
running, _ := buildstep.EncodeStep(buildstep.Step{ID: "mvl0frv18d1958", Tool: "build", Node: "ace", State: buildstep.StepRunning})
|
|
var mu sync.Mutex
|
|
var opens []uint64
|
|
var unanswered, answered int
|
|
asks := JetStreamAsks{
|
|
Unanswered: func(string, string, error) { mu.Lock(); unanswered++; mu.Unlock() },
|
|
Answered: func(string, string) { mu.Lock(); answered++; mu.Unlock() },
|
|
OpenWatch: func(_ context.Context, _, _ string, after uint64) (jetstream.KeyWatcher, error) {
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
opens = append(opens, after)
|
|
ch := make(chan jetstream.KeyValueEntry, 2)
|
|
// The bus answers nothing past the kept call's own asking again (its bound), so the fault is told and the watch tried again.
|
|
switch n := len(opens); {
|
|
case n <= 40:
|
|
return nil, nats.ErrNoResponders
|
|
case n == 41:
|
|
ch <- fakeEntry{rev: 5, value: running}
|
|
close(ch) // the watch closes under the wait
|
|
default:
|
|
ch <- fakeEntry{rev: 6, value: passed}
|
|
}
|
|
return fakeWatcher{updates: ch}, nil
|
|
}}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
steps, err := asks.WatchAsk(ctx, "mesh-build-go", "mvl0frv18d1958")
|
|
if err != nil {
|
|
t.Fatalf("a transport fault opening the watch failed the wait: %v", err)
|
|
}
|
|
var states []string
|
|
for s := range steps {
|
|
states = append(states, s.State)
|
|
if buildstep.Ended(s) {
|
|
break
|
|
}
|
|
}
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
if len(states) != 2 || states[0] != buildstep.StepRunning || states[1] != buildstep.StepPassed {
|
|
t.Fatalf("steps heard = %v, want running then passed", states)
|
|
}
|
|
if len(opens) != 42 || opens[41] != 6 {
|
|
t.Errorf("opens after revisions %v, want the last taken up again after the last revision heard (6)", opens)
|
|
}
|
|
if unanswered == 0 || answered != 2 {
|
|
t.Errorf("unanswered %d, answered %d, want the fault said and each answer said", unanswered, answered)
|
|
}
|
|
}
|