Merge pull request 'claude-code: the memory watch only notes a change, so it starts on a node with hundreds of files (issue 367)' (#279) from fix/367-memory-watch-takes-fast into main

2026-10-13 14:58:35 +00:00
parent 4d15a155e6 e46eeb8298
commit e209d46faf
2 changed files with 124 additions and 7 deletions
@@ -118,7 +118,11 @@ type MemoryCarrier struct {
 	Now    func() time.Time
 	Say    func(format string, args ...any)

-	mu         sync.Mutex
+	mu sync.Mutex
+	// queue is what the watch handed over and Drain has not applied yet: each key's newest revision.
+	queueMu    sync.Mutex
+	queue      map[string]uint64
+	wake       chan struct{}
 	rec        memoryRecord
 	loaded     bool
 	notCarried map[string]string
@@ -143,7 +147,7 @@ func HomeKind(p Paths) (string, string) {
 func NewMemoryCarrier(p Paths, store MemoryStore, notify func(map[string]any) error) *MemoryCarrier {
 	kind, why := HomeKind(p)
 	return &MemoryCarrier{Home: p.Home, Node: p.Node, State: p.State, Kind: kind, Why: why, Store: store,
-		Notify: notify, Now: time.Now, Say: say}
+		Notify: notify, Now: time.Now, Say: say, wake: make(chan struct{}, 1)}
 }

 func (m *MemoryCarrier) projects() string { return filepath.Join(m.Home, ".claude", "projects") }
@@ -894,10 +898,53 @@ func (m *MemoryCarrier) Take(key, op string, value json.RawMessage, revision uin
 	if err := badName(v.Project, v.Path); err != nil {
 		return fmt.Errorf("the state's value for %s is not taken: %w", key, err)
 	}
+	// Only noted here, never applied: the runtime hands the watch's values over one at a time and gives the whole
+	// current set ten seconds, so a watch whose handler reads the state, waits for a look or writes a file never
+	// starts on a node with a few hundred memory files (issue 367, seen on shanks on 2026-10-13). Drain applies
+	// what was noted, the newest revision of each key.
+	m.queueMu.Lock()
+	if m.queue == nil {
+		m.queue = map[string]uint64{}
+	}
+	if m.queue[key] < revision {
+		m.queue[key] = revision
+	}
+	m.queueMu.Unlock()
+	select {
+	case m.wake <- struct{}{}:
+	default:
+	}
+	return nil
+}
+
+// Drain applies every change Take noted, each key once, at the state's current version (ADR 0343 §6).
+func (m *MemoryCarrier) Drain() error {
+	m.queueMu.Lock()
+	queued := m.queue
+	m.queue = nil
+	m.queueMu.Unlock()
+	if len(queued) == 0 {
+		return nil
+	}
+	keys := make([]string, 0, len(queued))
+	for k := range queued {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	m.load()
-	// A change handed over late is never written over a newer one: what is applied is the state's current version.
+	var problems []error
+	for _, key := range keys {
+		if err := m.takeNow(key, queued[key]); err != nil {
+			problems = append(problems, err)
+		}
+	}
+	return errors.Join(append(problems, m.save())...)
+}
+
+// takeNow applies one key at the state's current version: a change noted late is never written over a newer one.
+func (m *MemoryCarrier) takeNow(key string, revision uint64) error {
 	cur, rev, err := m.current(key)
 	if err != nil {
 		return err
@@ -908,8 +955,7 @@ func (m *MemoryCarrier) Take(key, op string, value json.RawMessage, revision uin
 	if cur.Kind != m.Kind || MemoryKey(m.Kind, cur.Project, cur.Path) != key {
 		return fmt.Errorf("the state's value for %s names another file or kind, and is not taken", key)
 	}
-	err = m.apply(key, *cur, rev)
-	return errors.Join(err, m.save())
+	return m.apply(key, *cur, rev)
 }

 // apply writes one incoming version, read at revision, into this node's folder (ADR 0343 §6, §7, §9).
@@ -1075,6 +1121,13 @@ func (m *MemoryCarrier) Run(watch func(handle func(key, op string, value json.Ra
 		}
 		return
 	}
+	go func() {
+		for range m.wake {
+			if err := m.Drain(); err != nil && m.Say != nil {
+				m.Say("memory: %v", err)
+			}
+		}
+	}()
 	go persist("watching the agent's memory", func() error {
 		return watch(func(key, op string, value json.RawMessage, revision uint64) error {
 			if err := m.Take(key, op, value, revision); err != nil && m.Say != nil {
@@ -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)
+	}
+}