claude-code: the memory watch only notes a change, so it starts on a node with hundreds of files (issue 367)

The runtime hands a watch its current values one at a time and gives them ten
seconds. The handler read the state, waited for a running look and wrote files
for each value, so with about 450 memory values the watch never started on g14
or shanks, and a file written on g14 never reached shanks. The handler now
notes each key's newest revision and returns; a worker applies what was noted
at the state's current version.

modules/claude-code/cmd/claude-code/memory_test.go

@@ -106,11 +106,16 @@ func (s *memState) pump(t *testing.T) {
 		}
 		for _, c := range todo {
 			for _, w := range s.watchers {
-				if err := w.Take(c.key, c.op, c.value, c.rev); err != nil && !errors.Is(err, errChanged) {
+				if err := w.Take(c.key, c.op, c.value, c.rev); err != nil {
 					t.Logf("%s taking %s: %v", w.Node, c.key, err)
 				}
 			}
 		}
+		for _, w := range s.watchers {
+			if err := w.Drain(); err != nil && !errors.Is(err, errChanged) {
+				t.Logf("%s applying: %v", w.Node, err)
+			}
+		}
 	}
 	t.Fatal("the state never settled")
 }
@@ -137,6 +142,7 @@ func newMemNode(t *testing.T, root, name, kind string, state *memState, c *clock
 		}
 	}
 	n.MemoryCarrier = &MemoryCarrier{Home: home, Node: name, State: stateDir, Kind: kind, Store: state, Now: c.Now,
+		wake:   make(chan struct{}, 1),
 		Notify: func(a map[string]any) error { n.notices = append(n.notices, a); return nil }}
 	state.watchers = append(state.watchers, n.MemoryCarrier)
 	return n
@@ -601,10 +607,13 @@ func (s *memState) replay(t *testing.T, n *memNode) {
 	keys, _ := s.Keys()
 	for _, k := range keys {
 		raw, rev, _, _ := s.Get(k)
-		if err := n.Take(k, "put", raw, rev); err != nil && !errors.Is(err, errChanged) {
+		if err := n.Take(k, "put", raw, rev); err != nil {
 			t.Logf("%s taking %s: %v", n.Node, k, err)
 		}
 	}
+	if err := n.Drain(); err != nil && !errors.Is(err, errChanged) {
+		t.Logf("%s applying: %v", n.Node, err)
+	}
 }
 
 // §6: a node that was away while a file changed twice takes the newest version, whose base it never saw, because
@@ -902,3 +911,58 @@ func TestARemovalReachesANodeThatWasAway(t *testing.T) {
 		t.Fatal("the version removed on the node that was away was not kept")
 	}
 }
+
+// slowState answers every read only once it is let go, as a state a whole bus away does under load.
+type slowState struct {
+	*memState
+	gate chan struct{}
+}
+
+func (s slowState) Get(key string) (json.RawMessage, uint64, bool, error) {
+	<-s.gate
+	return s.memState.Get(key)
+}
+
+// Issue 367, seen live: the runtime hands a watch its current values one at a time and gives them all ten seconds;
+// shanks's handler read the state and waited for its look for each of a few hundred values, so the watch never
+// started and nothing written on g14 reached it. The handler now only notes a change: it returns at once while a
+// look holds the carrier and while the state is slow to answer, and Drain applies what it noted.
+func TestTheWatchHandlerReturnsAtOnceAndDrainAppliesWhatItNoted(t *testing.T) {
+	state, c, g14, shanks := twoNodes(t)
+	g14.write(t, project, "MEMORY.md", "index\n", c.now)
+	g14.look(t)
+	state.mu.Lock()
+	state.pending = nil
+	state.mu.Unlock()
+	slow := slowState{memState: state, gate: make(chan struct{})}
+	shanks.Store = slow
+	key := MemoryKey(HomeOperator, project, "MEMORY.md")
+	raw, rev, _, _ := state.Get(key)
+	shanks.mu.Lock() // a look running
+	done := make(chan error, 1)
+	start := time.Now()
+	for i := 0; i < 400; i++ {
+		go func() { done <- shanks.Take(key, "put", raw, rev) }()
+	}
+	for i := 0; i < 400; i++ {
+		select {
+		case err := <-done:
+			if err != nil {
+				t.Fatal(err)
+			}
+		case <-time.After(2 * time.Second):
+			t.Fatalf("the watch handler waited for the look or the state (%d of 400 returned)", i)
+		}
+	}
+	if took := time.Since(start); took > time.Second {
+		t.Fatalf("400 values took %s to hand over", took)
+	}
+	shanks.mu.Unlock()
+	close(slow.gate)
+	if err := shanks.Drain(); err != nil {
+		t.Fatal(err)
+	}
+	if got, _ := shanks.read(t, project, "MEMORY.md"); got != "index\n" {
+		t.Fatalf("what the watch noted was not applied: %q", got)
+	}
+}