From 1fc006b7ddfa6869cc1cac2915fd3dc37ae6616a Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 02:17:35 +0100 Subject: [PATCH 1/2] The runtime says a state or log operation of a bundle that failed or timed out, with its time (issue 1966) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- node-tools/internal/launch/launch.go | 8 +++ .../internal/launch/state_failed_test.go | 64 +++++++++++++++++++ 2 files changed, 72 insertions(+) create mode 100644 node-tools/internal/launch/state_failed_test.go diff --git a/node-tools/internal/launch/launch.go b/node-tools/internal/launch/launch.go index 0044ea5..bcb593c 100644 --- a/node-tools/internal/launch/launch.go +++ b/node-tools/internal/launch/launch.go @@ -345,6 +345,7 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an go func(m message) { var result any = map[string]any{} var err error + asked := time.Now() switch m.Method { case "mesh/publish": err = mesh.Publish(m.Params) @@ -479,6 +480,13 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an } return } + // A state or log operation the bus refused or let time out reaches a caller as a call that + // does not answer: say it here, with how long it took, so the next hang is readable. + if err != nil && (strings.HasPrefix(m.Method, "mesh/state.") || strings.HasPrefix(m.Method, "mesh/log.") || + busStateCalls[m.Method]) { + logf("[mesh-tools] %s's bundle asked %s and it failed after %s: %v", module, m.Method, + time.Since(asked).Round(time.Millisecond), err) + } if len(m.ID) == 0 { return } diff --git a/node-tools/internal/launch/state_failed_test.go b/node-tools/internal/launch/state_failed_test.go new file mode 100644 index 0000000..7ee0f99 --- /dev/null +++ b/node-tools/internal/launch/state_failed_test.go @@ -0,0 +1,64 @@ +package launch + +import ( + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" +) + +type refusingState struct{ subscribing } + +func (*refusingState) State(string, json.RawMessage) (json.RawMessage, error) { + return nil, errors.New("the bus did not answer in 10s") +} + +// A bundle that asks the runtime for a state once it is started, as a tool does. +const askingBundle = `#!/bin/sh +while IFS= read -r line; do + id=$(printf '%s' "$line" | sed -n 's/.*"id":\([0-9]*\)[,}].*/\1/p') + case "$line" in + *'"method":"initialize"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" ;; + *'"method":"tools/list"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{"tools":[]}}\n' "$id" + printf '{"jsonrpc":"2.0","id":"q1","method":"mesh/state.get","params":{"state":"copied","key":"k"}}\n' ;; + esac +done +` + +// A state operation the bus refuses or lets time out is said in the runtime's log with its method, its bundle and its +// time (issue 1966: a hand-act seat whose calls hung said nothing). +func TestAFailedStateOperationIsSaidInTheRuntimesLog(t *testing.T) { + dir := t.TempDir() + entry := filepath.Join(dir, "bundle") + if err := os.WriteFile(entry, []byte(askingBundle), 0o755); err != nil { + t.Fatal(err) + } + var mu sync.Mutex + var lines []string + logf := func(f string, a ...any) { + mu.Lock() + defer mu.Unlock() + lines = append(lines, fmt.Sprintf(f, a...)) + } + l, err := Start("asker", entry, os.Environ(), &refusingState{}, logf) + if err != nil { + t.Fatal(err) + } + t.Cleanup(l.Stop) + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + mu.Lock() + all := strings.Join(lines, "\n") + mu.Unlock() + if strings.Contains(all, "asker's bundle asked mesh/state.get and it failed after") && strings.Contains(all, "did not answer in 10s") { + return + } + time.Sleep(20 * time.Millisecond) + } + t.Fatalf("the failed state operation was not said; the log: %v", lines) +} -- 2.54.0 From 634c77f90f9813746d58d1d5b83361c5415dd858 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 02:21:55 +0100 Subject: [PATCH 2/2] The runtime says no lost compare-and-set as a failed operation, and its test waits on a channel, never a sleep (review round 1) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- node-tools/internal/launch/launch.go | 5 +- .../internal/launch/state_failed_test.go | 53 +++++++++++-------- 2 files changed, 35 insertions(+), 23 deletions(-) diff --git a/node-tools/internal/launch/launch.go b/node-tools/internal/launch/launch.go index bcb593c..69b0678 100644 --- a/node-tools/internal/launch/launch.go +++ b/node-tools/internal/launch/launch.go @@ -482,8 +482,9 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an } // A state or log operation the bus refused or let time out reaches a caller as a call that // does not answer: say it here, with how long it took, so the next hang is readable. - if err != nil && (strings.HasPrefix(m.Method, "mesh/state.") || strings.HasPrefix(m.Method, "mesh/log.") || - busStateCalls[m.Method]) { + // A lost compare-and-set is the normal answer of two writers to one key, retried by its caller: not said. + if err != nil && errorCode(err) != CodeChanged && (strings.HasPrefix(m.Method, "mesh/state.") || + strings.HasPrefix(m.Method, "mesh/log.") || busStateCalls[m.Method]) { logf("[mesh-tools] %s's bundle asked %s and it failed after %s: %v", module, m.Method, time.Since(asked).Round(time.Millisecond), err) } diff --git a/node-tools/internal/launch/state_failed_test.go b/node-tools/internal/launch/state_failed_test.go index 7ee0f99..1b9b7ff 100644 --- a/node-tools/internal/launch/state_failed_test.go +++ b/node-tools/internal/launch/state_failed_test.go @@ -7,58 +7,69 @@ import ( "os" "path/filepath" "strings" - "sync" "testing" "time" ) +// changedError is a lost compare-and-set as the bus answers it: told by its code (novox/hq ADR 0259). +type changedError struct{} + +func (changedError) Error() string { return "the key changed since revision 3" } +func (changedError) ErrorCode() int { return CodeChanged } + +// refusingState answers the state "lost" with a lost compare-and-set and every other state with a time-out. type refusingState struct{ subscribing } -func (*refusingState) State(string, json.RawMessage) (json.RawMessage, error) { +func (*refusingState) State(_ string, params json.RawMessage) (json.RawMessage, error) { + if strings.Contains(string(params), `"state":"lost"`) { + return nil, changedError{} + } return nil, errors.New("the bus did not answer in 10s") } -// A bundle that asks the runtime for a state once it is started, as a tool does. +// A bundle that asks the runtime for the state "lost" once it is started, and for the state "copied" once that is +// answered, so the first answer's handling has ended before the second is asked. const askingBundle = `#!/bin/sh while IFS= read -r line; do id=$(printf '%s' "$line" | sed -n 's/.*"id":\([0-9]*\)[,}].*/\1/p') case "$line" in *'"method":"initialize"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" ;; *'"method":"tools/list"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{"tools":[]}}\n' "$id" - printf '{"jsonrpc":"2.0","id":"q1","method":"mesh/state.get","params":{"state":"copied","key":"k"}}\n' ;; + printf '{"jsonrpc":"2.0","id":"q1","method":"mesh/state.update","params":{"state":"lost","key":"k"}}\n' ;; + *'"id":"q1"'*) printf '{"jsonrpc":"2.0","id":"q2","method":"mesh/state.get","params":{"state":"copied","key":"k"}}\n' ;; esac done ` // A state operation the bus refuses or lets time out is said in the runtime's log with its method, its bundle and its -// time (issue 1966: a hand-act seat whose calls hung said nothing). +// time (issue 1966: a hand-act seat whose calls hung said nothing); a lost compare-and-set, asked first, is not. func TestAFailedStateOperationIsSaidInTheRuntimesLog(t *testing.T) { dir := t.TempDir() entry := filepath.Join(dir, "bundle") if err := os.WriteFile(entry, []byte(askingBundle), 0o755); err != nil { t.Fatal(err) } - var mu sync.Mutex - var lines []string - logf := func(f string, a ...any) { - mu.Lock() - defer mu.Unlock() - lines = append(lines, fmt.Sprintf(f, a...)) - } + said := make(chan string, 16) + logf := func(f string, a ...any) { said <- fmt.Sprintf(f, a...) } l, err := Start("asker", entry, os.Environ(), &refusingState{}, logf) if err != nil { t.Fatal(err) } t.Cleanup(l.Stop) - deadline := time.Now().Add(5 * time.Second) - for time.Now().Before(deadline) { - mu.Lock() - all := strings.Join(lines, "\n") - mu.Unlock() - if strings.Contains(all, "asker's bundle asked mesh/state.get and it failed after") && strings.Contains(all, "did not answer in 10s") { - return + var heard []string + timeout := time.After(5 * time.Second) + for { + select { + case line := <-said: + heard = append(heard, line) + if strings.Contains(line, "mesh/state.update") { + t.Fatalf("a lost compare-and-set was said as a failure: %q", line) + } + if strings.Contains(line, "asker's bundle asked mesh/state.get and it failed after") && strings.Contains(line, "did not answer in 10s") { + return + } + case <-timeout: + t.Fatalf("the failed state operation was not said; the log: %v", heard) } - time.Sleep(20 * time.Millisecond) } - t.Fatalf("the failed state operation was not said; the log: %v", lines) } -- 2.54.0