mesh-tools/node-tools/internal/launch/state_failed_test.go

package launch

import (
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"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, 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 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.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); 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)
	}
	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)
	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)
		}
	}
}

// A bundle that lists one tool, "t", and answers a call to it as its MODE says: "long" with a line over the runner's cap,
// "silent" with nothing at all.
const callingBundle = `#!/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"'*) echo started >&2; printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" ;;
    *'"method":"tools/list"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{"tools":[{"name":"t","description":"d"}]}}\n' "$id" ;;
    *'"method":"tools/call"'*) [ "$MODE" = long ] && printf '%4096s\n' x ;;
  esac
done
`

// Issue 1966: a bundle's line over the runner's cap, and a call the bundle never answers, are said in the runtime's log;
// the over-long line ends the bundle and fails the call at once, never leaving the runner reading nothing while the
// bundle's next write blocks.
func TestAnOverlongLineAndAnUnansweredCallAreSaid(t *testing.T) {
	oldMax, oldCall, oldFirst := maxLine, CallTimeout, RestartFirst
	maxLine, RestartFirst = 1024, 10*time.Millisecond
	t.Cleanup(func() { maxLine, CallTimeout, RestartFirst = oldMax, oldCall, oldFirst })
	for _, row := range []struct {
		mode    string
		timeout time.Duration
	}{
		// The call's own bound is long: only the runner ending the bundle fails it within a second.
		{"long", 10 * time.Second},
		{"silent", 300 * time.Millisecond},
	} {
		t.Run(row.mode, func(t *testing.T) {
			CallTimeout = row.timeout
			dir := t.TempDir()
			entry := filepath.Join(dir, "bundle")
			if err := os.WriteFile(entry, []byte(callingBundle), 0o755); err != nil {
				t.Fatal(err)
			}
			said := make(chan string, 32)
			logf := func(f string, a ...any) { said <- fmt.Sprintf(f, a...) }
			l, err := Start("caller", entry, append(os.Environ(), "MODE="+row.mode), &refusingState{}, logf)
			if err != nil {
				t.Fatal(err)
			}
			t.Cleanup(l.Stop)
			begun := time.Now()
			_, err = l.Registrations[0].Tools[0].Run("caller-1", nil)
			if err == nil {
				t.Fatal("the call was answered")
			}
			var heard []string
			timeout := time.After(5 * time.Second)
			want := "caller.t called by \"caller-1\""
			if row.mode == "long" {
				want = "ending the bundle"
				if took := time.Since(begun); took > time.Second {
					t.Fatalf("the call hung for %s", took)
				}
				if !strings.Contains(err.Error(), "over 1024 bytes") || strings.Contains(err.Error(), "did not answer") {
					t.Fatalf("the call failed with %q, not the cap's reason", err)
				}
			}
			// For "long" the bundle is also started again: it says "started" on each start, so twice in all.
			started, found := 0, false
			for !found || (row.mode == "long" && started < 2) {
				select {
				case line := <-said:
					heard = append(heard, line)
					found = found || strings.Contains(line, want)
					if line == "[caller] started" {
						started++
					}
				case <-timeout:
					t.Fatalf("not said, or the bundle not started again; the log: %v", heard)
				}
			}
		})
	}
}