S4: the assigner's wants step proves a restatement heard while ace's ask runs is asked once more (issue 1966 note 170)
mesh/delivery merged: source.main moved to novox/mesh-delivery@a6bba2af808c
mesh/delivery-group group fix/1966-engine-reads-asked-again merged: every member is merged
mesh/check checking 1a6144aa352b on a6bba2af808c | novox/mesh-delivery#201 in group fix/1966-engine-reads-asked-again
mesh/delivery merged: source.main moved to novox/mesh-delivery@a6bba2af808c
mesh/delivery-group group fix/1966-engine-reads-asked-again merged: every member is merged
mesh/check checking 1a6144aa352b on a6bba2af808c | novox/mesh-delivery#201 in group fix/1966-engine-reads-asked-again
1966/AC170-4: in TestWantsAreWrittenAndNoDifferenceIs, ace's engine's answer is held while it restates with a new form and says manifest-reads-changed again; the restatement's event is asked once more after the held ask (the journal line), the state engine-reads ends at the restatement with its event-at, and the held want moves. Against the node-engine seat's client on trunk the step fails: the event is dropped. A step in the existing test, no new function (the repository is over its test budget); the bed hands the program a journal the test awaits a line in, no sleep. Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
@@ -69,6 +69,52 @@ type bed struct {
|
||||
buckets map[string]jetstream.KeyValue
|
||||
// placed is, by node, the modules the bed's assignments names on it (place).
|
||||
placed map[string][]string
|
||||
// journal is what the programs the bed starts write in their journal.
|
||||
journal *journal
|
||||
}
|
||||
|
||||
// journal is a program's journal as the bed hears it: a test awaits a line in it (await) where the line is the only sign
|
||||
// that the program took something in; each write wakes the waiter, no sleep stands in for it.
|
||||
type journal struct {
|
||||
mu sync.Mutex
|
||||
text strings.Builder
|
||||
grew chan struct{}
|
||||
}
|
||||
|
||||
func (j *journal) Write(p []byte) (int, error) {
|
||||
j.mu.Lock()
|
||||
defer j.mu.Unlock()
|
||||
j.text.Write(p)
|
||||
close(j.grew)
|
||||
j.grew = make(chan struct{})
|
||||
return len(p), nil
|
||||
}
|
||||
|
||||
// mark is where the journal ends now, for await.
|
||||
func (j *journal) mark() int {
|
||||
j.mu.Lock()
|
||||
defer j.mu.Unlock()
|
||||
return j.text.Len()
|
||||
}
|
||||
|
||||
// await returns once the journal holds, after from (mark), a line with words, and fails the test when none came within
|
||||
// bound.
|
||||
func (j *journal) await(t *testing.T, from int, words, what string, bound time.Duration) {
|
||||
t.Helper()
|
||||
deadline := time.After(bound)
|
||||
for {
|
||||
j.mu.Lock()
|
||||
has, grew := strings.Contains(j.text.String()[from:], words), j.grew
|
||||
j.mu.Unlock()
|
||||
if has {
|
||||
return
|
||||
}
|
||||
select {
|
||||
case <-grew:
|
||||
case <-deadline:
|
||||
t.Fatalf("%s: no line %q in the journal within %s", what, words, bound)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func startBed(t *testing.T, leaveOut string) *bed {
|
||||
@@ -98,7 +144,7 @@ func startBed(t *testing.T, leaveOut string) *bed {
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||||
t.Cleanup(cancel)
|
||||
b := &bed{srv: srv, js: js, ctx: ctx, dir: dir, buckets: map[string]jetstream.KeyValue{}}
|
||||
b := &bed{srv: srv, js: js, ctx: ctx, dir: dir, buckets: map[string]jetstream.KeyValue{}, journal: &journal{grew: make(chan struct{})}}
|
||||
for _, name := range append(append([]string{}, readBuckets...), ownBuckets...) {
|
||||
if name == leaveOut {
|
||||
continue
|
||||
@@ -170,7 +216,7 @@ func (b *bed) startStoppable(t *testing.T) (<-chan error, func()) {
|
||||
ctx, stop := context.WithCancel(b.ctx)
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
done <- run(ctx, func(k string) string { return env[k] }, io.Discard)
|
||||
done <- run(ctx, func(k string) string { return env[k] }, b.journal)
|
||||
close(done) // a program that ended on its own (a refused start) is not waited for again at the cleanup
|
||||
}()
|
||||
stopped := func() {
|
||||
@@ -596,6 +642,9 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) {
|
||||
// subject and states the form; no timer moves it. g14's engine answers nothing and has no record: read as the baseline.
|
||||
stated := catalogue.ManifestReads()
|
||||
var statedMu sync.Mutex
|
||||
// hold, when set, holds the next ask of ace's engine, answered with what it stated when asked, until it is closed;
|
||||
// held is closed once that ask is held (issue 1966 note 170).
|
||||
var hold, held chan struct{}
|
||||
nc, err := nats.Connect(b.srv.ClientURL())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -604,7 +653,13 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) {
|
||||
if _, err := nc.Subscribe("mesh.seat.node-engine.tool.manifest-reads.ace", func(m *nats.Msg) {
|
||||
statedMu.Lock()
|
||||
answer, _ := json.Marshal(nodeengine.ManifestReadsAnswer{V: 1, Node: "ace", ManifestReads: stated, StatedAt: time.Now().Format(time.RFC3339Nano)})
|
||||
release, isHeld := hold, held
|
||||
hold = nil
|
||||
statedMu.Unlock()
|
||||
if release != nil {
|
||||
close(isHeld)
|
||||
<-release
|
||||
}
|
||||
reply, _ := json.Marshal(map[string]any{"result": json.RawMessage(answer), "node": "ace"})
|
||||
_ = m.Respond(reply)
|
||||
}); err != nil {
|
||||
@@ -669,10 +724,31 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) {
|
||||
}); w.Fp != formsOld || w.Rule != mwc.RuleComposed {
|
||||
t.Errorf("forms' want = %+v, want it held at what ace runs under rule 7", w)
|
||||
}
|
||||
// Issue 1966 note 170, 1966/AC170-4: ace's engine says its statement, and while the assigner's ask of it is held,
|
||||
// restates it with the form and says that too; the ask held answers the first statement. The restatement's event,
|
||||
// heard while that ask ran, is asked once more after it, said in the journal, so the state ends at the restatement
|
||||
// and the held want moves.
|
||||
release := make(chan struct{})
|
||||
statedMu.Lock()
|
||||
hold, held = release, make(chan struct{})
|
||||
isHeld := held
|
||||
statedMu.Unlock()
|
||||
say(enginereads.ReadsStarted)
|
||||
select {
|
||||
case <-isHeld:
|
||||
case <-time.After(15 * time.Second):
|
||||
t.Fatal("the assigner did not ask ace's engine on its manifest-reads-changed")
|
||||
}
|
||||
from := b.journal.mark()
|
||||
statedMu.Lock()
|
||||
stated = append(slices.Clone(stated), "form:new")
|
||||
statedMu.Unlock()
|
||||
say(enginereads.ReadsStarted)
|
||||
b.journal.await(t, from, "manifest-reads on ace: an ask runs, so it is asked once more after it", "the restatement heard while ace was asked", 15*time.Second)
|
||||
close(release)
|
||||
if s := statement("ace", func(s enginereads.Statement) bool { return slices.Contains(s.ManifestReads, "form:new") }); s.EventAt.IsZero() {
|
||||
t.Errorf("ace's statement does not show when its event was heard: %+v", s)
|
||||
}
|
||||
b.wantNow(t, "ace", "forms", "the want moved once ace's engine said it reads the form", func(w mwc.Want) bool { return w.Fp == formsNew })
|
||||
// assigner_engine_reads asks ace's engine again through the state: the assigner answers there with what it states now.
|
||||
again, err := enginereads.AskAgainOf(bedState{b, engineReadsBucket}, "ace", "n1.ada", "a person asks", time.Now(), 15*time.Second)
|
||||
|
||||
Reference in New Issue
Block a user