From e46eeb829853c4b6a9704d3747554338fbac1392 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 13 Oct 2026 16:40:41 +0200 Subject: [PATCH] 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.go | 63 +++++++++++++++-- .../cmd/claude-code/memory_test.go | 68 ++++++++++++++++++- 2 files changed, 124 insertions(+), 7 deletions(-) diff --git a/modules/claude-code/cmd/claude-code/memory.go b/modules/claude-code/cmd/claude-code/memory.go index ad413727..0315301d 100644 --- a/modules/claude-code/cmd/claude-code/memory.go +++ b/modules/claude-code/cmd/claude-code/memory.go @@ -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 { diff --git a/modules/claude-code/cmd/claude-code/memory_test.go b/modules/claude-code/cmd/claude-code/memory_test.go index 48068835..892e8f95 100644 --- a/modules/claude-code/cmd/claude-code/memory_test.go +++ b/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) + } +}