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
|
buckets map[string]jetstream.KeyValue
|
||||||
// placed is, by node, the modules the bed's assignments names on it (place).
|
// placed is, by node, the modules the bed's assignments names on it (place).
|
||||||
placed map[string][]string
|
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 {
|
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)
|
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||||||
t.Cleanup(cancel)
|
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...) {
|
for _, name := range append(append([]string{}, readBuckets...), ownBuckets...) {
|
||||||
if name == leaveOut {
|
if name == leaveOut {
|
||||||
continue
|
continue
|
||||||
@@ -170,7 +216,7 @@ func (b *bed) startStoppable(t *testing.T) (<-chan error, func()) {
|
|||||||
ctx, stop := context.WithCancel(b.ctx)
|
ctx, stop := context.WithCancel(b.ctx)
|
||||||
done := make(chan error, 1)
|
done := make(chan error, 1)
|
||||||
go func() {
|
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
|
close(done) // a program that ended on its own (a refused start) is not waited for again at the cleanup
|
||||||
}()
|
}()
|
||||||
stopped := func() {
|
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.
|
// 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()
|
stated := catalogue.ManifestReads()
|
||||||
var statedMu sync.Mutex
|
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())
|
nc, err := nats.Connect(b.srv.ClientURL())
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
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) {
|
if _, err := nc.Subscribe("mesh.seat.node-engine.tool.manifest-reads.ace", func(m *nats.Msg) {
|
||||||
statedMu.Lock()
|
statedMu.Lock()
|
||||||
answer, _ := json.Marshal(nodeengine.ManifestReadsAnswer{V: 1, Node: "ace", ManifestReads: stated, StatedAt: time.Now().Format(time.RFC3339Nano)})
|
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()
|
statedMu.Unlock()
|
||||||
|
if release != nil {
|
||||||
|
close(isHeld)
|
||||||
|
<-release
|
||||||
|
}
|
||||||
reply, _ := json.Marshal(map[string]any{"result": json.RawMessage(answer), "node": "ace"})
|
reply, _ := json.Marshal(map[string]any{"result": json.RawMessage(answer), "node": "ace"})
|
||||||
_ = m.Respond(reply)
|
_ = m.Respond(reply)
|
||||||
}); err != nil {
|
}); err != nil {
|
||||||
@@ -669,10 +724,31 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) {
|
|||||||
}); w.Fp != formsOld || w.Rule != mwc.RuleComposed {
|
}); w.Fp != formsOld || w.Rule != mwc.RuleComposed {
|
||||||
t.Errorf("forms' want = %+v, want it held at what ace runs under rule 7", w)
|
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()
|
statedMu.Lock()
|
||||||
stated = append(slices.Clone(stated), "form:new")
|
stated = append(slices.Clone(stated), "form:new")
|
||||||
statedMu.Unlock()
|
statedMu.Unlock()
|
||||||
say(enginereads.ReadsStarted)
|
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 })
|
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.
|
// 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)
|
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