diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 4b845e2..c692857 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -43,4 +43,10 @@ jobs: run: make install - name: Compile - run: make compile \ No newline at end of file + run: make compile + + - name: Test + run: make test + + - name: Race-test execution paths + run: go test -race ./pkg/runtime/... ./pkg/run/... ./pkg/repl/... ./cmd/... ./internal/diagnostics/... diff --git a/Makefile b/Makefile index 577e798..072323e 100644 --- a/Makefile +++ b/Makefile @@ -27,7 +27,7 @@ test: fmt: go fmt ./... && \ - goimports -w -local github.com/MontFerret ./internal ./pkg/browser ./cmd ./pkg/debugger ./pkg/config ./ferret ./pkg/logger ./pkg/module ./pkg/repl ./pkg/runtime + goimports -w -local github.com/MontFerret ./internal ./pkg/browser ./cmd ./pkg/debugger ./pkg/config ./ferret ./pkg/logger ./pkg/module ./pkg/repl ./pkg/runtime ./examples lint: staticcheck ./... && \ diff --git a/README.md b/README.md index 3641367..7ff8278 100644 --- a/README.md +++ b/README.md @@ -426,6 +426,84 @@ and an exact open pull request or publication branch is reused. Ferret never overwrites an immutable Registry record or a divergent branch; delete a stale publication branch from your fork before retrying if its contents differ. +## Wire source execution + +This CLI integrates Wire `v1.0.0-alpha.2`, Universal API `v1.0.0-alpha.20`, and +Ferret `v2.0.0-alpha.57`. The host must support the alpha.2 handshake, including +hosted runtime version metadata; alpha.1 hosts are incompatible. + +Builtin execution remains the default. Select `--runtime wire` explicitly to run +source against an application-configured Universal API runtime. Legacy HTTP URLs +in `--runtime` keep their existing Worker behavior. + +Start the development host in terminal 1: + +```bash +go run ./examples/wire-host +``` + +It prints an ephemeral endpoint such as `tcp://127.0.0.1:54321`. In terminal 2, +substitute that endpoint in these commands: + +```bash +ferret run --runtime wire --runtime-endpoint tcp://127.0.0.1:54321 \ + --eval 'RETURN DEMO::IDENTITY(@value)' --param value=42 +ferret exec --runtime wire --runtime-endpoint tcp://127.0.0.1:54321 script.fql +ferret run --runtime wire --runtime-endpoint tcp://127.0.0.1:54321 \ + --eval 'let arr = [] + +return' +ferret repl --runtime wire --runtime-endpoint tcp://127.0.0.1:54321 +ferret version --runtime wire --runtime-endpoint tcp://127.0.0.1:54321 +``` + +`DEMO::IDENTITY(value)` exists only in the example host. Try it in the REPL, submit +`RETURN )`, then correct the query. Syntax errors retain source snippets, carets, +and hints. Use `exit` or Ctrl-D to exit; Ctrl-C stops the shell. Stop the example +host with Ctrl-C when finished. + +| Flag and config key | Environment variable | Default | +| --- | --- | --- | +| `runtime-endpoint` | `FERRET_RUNTIME_ENDPOINT` | Required for Wire; no default port | +| `runtime-connect-timeout` | `FERRET_RUNTIME_CONNECT_TIMEOUT` | `5s`, must be positive | + +Only `tcp://127.0.0.1:` is supported (ports 1–65535, no other URL components). +Both connection establishment and the Wire handshake respect the timeout and +command cancellation. Connection failure never falls back or reconnects the +logical runtime automatically. Endpoint and connection-timeout settings require +Wire mode; builtin and legacy HTTP behavior otherwise remain unchanged. + +This plaintext transport is for **trusted local development**. Loopback access +does not authenticate either the client or runtime. There are no TLS or credential +settings in this integration. HTTP headers and policy flags are not Wire credentials. + +The CLI reads script files locally and transmits their semantic name and contents. +Compilation happens on the host, allowing application-only functions and modules. +FQL filesystem operations use the runtime host's configured filesystem defaults +(the example host uses its working directory as the root); +the CLI does not send its current directory as a filesystem root. Explicit +filesystem/HTTP policies and browser configuration, including settings in config +or environment variables, are rejected in Wire mode. Untouched flag defaults do +not count as explicit configuration. + +Each run owns one logical runtime and transport; the REPL reuses one pair. Active +calls settle before runtime and transport cleanup. Wire performs detached cleanup +bounded by its 30-second cleanup timeout after cancellation. The host owns its +engine independently; closing a CLI connection does not close it. + +Run output is printed as encoded bytes, including available output accompanying +an error. Errors go to stderr and fail the command; successfully produced empty +output prints nothing. Artifacts and debugging remain builtin-only. + +`ferret version --runtime wire` reports the hosted runtime version alongside the +CLI version. The Wire REPL banner shows the endpoint and hosted version. Version +values come directly from the Universal API and are preserved as opaque strings, +including empty values. Wire captures this metadata during the initial handshake; +reading it adds no RPC. Hosted metadata failure fails construction within the +connection timeout, before the REPL requests input. The example derives its +Ferret version from Go build information; builtin execution retains CLI-injected +version metadata (or `unknown` in a build without version injection). + ## Browser usage Ferret can use Chrome or Chromium through the Chrome DevTools Protocol. diff --git a/cmd/internal/debug/debug_test.go b/cmd/internal/debug/debug_test.go index 87db83f..0d72483 100644 --- a/cmd/internal/debug/debug_test.go +++ b/cmd/internal/debug/debug_test.go @@ -8,6 +8,7 @@ import ( "github.com/MontFerret/cli/v2/cmd/internal/execution" "github.com/MontFerret/cli/v2/cmd/internal/testutil" + "github.com/MontFerret/cli/v2/internal/testutil/wirehost" "github.com/MontFerret/ferret/v2/pkg/source" "github.com/MontFerret/cli/v2/pkg/browser" @@ -30,6 +31,21 @@ func TestDebugCommandRequiresExactlyOneScript(t *testing.T) { } } +func TestDebugRejectsWireBeforeConnecting(t *testing.T) { + host := wirehost.New(t) + path := filepath.Join(t.TempDir(), "script.fql") + testutil.WriteQuery(t, path, "RETURN DEMO::IDENTITY(42)") + store := testutil.NewStore(t) + cmd := New(store) + cmd.SilenceErrors = true + cmd.SilenceUsage = true + cmd.SetArgs([]string{"--runtime=wire", "--runtime-endpoint=" + host.Endpoint, path}) + err := cmd.ExecuteContext(config.With(t.Context(), store)) + if !errors.Is(err, cliruntime.ErrDebugRequiresBuiltinRuntime) || host.Connections() != 0 { + t.Fatalf("error=%v connections=%d", err, host.Connections()) + } +} + func TestExecuteDebugRejectsRemoteRuntimeBeforeStarting(t *testing.T) { path := filepath.Join(t.TempDir(), "query.fql") testutil.WriteQuery(t, path, "RETURN 1") diff --git a/cmd/internal/diagnostics/diagnostics.go b/cmd/internal/diagnostics/diagnostics.go index c62d177..28b6741 100644 --- a/cmd/internal/diagnostics/diagnostics.go +++ b/cmd/internal/diagnostics/diagnostics.go @@ -1,13 +1,18 @@ package diagnostics import ( - "fmt" + "io" "os" - "github.com/MontFerret/ferret/v2/pkg/diagnostics" + "github.com/MontFerret/cli/v2/internal/diagnostics" ) // PrintError preserves the CLI's direct stderr rendering for Ferret diagnostics. func PrintError(err error) { - fmt.Fprintln(os.Stderr, diagnostics.Format(err)) + diagnostics.Print(os.Stderr, err) +} + +// PrintErrorTo renders execution diagnostics to the command's selected stderr. +func PrintErrorTo(out io.Writer, err error) { + diagnostics.Print(out, err) } diff --git a/cmd/internal/execution/flags.go b/cmd/internal/execution/flags.go index 052fb21..0c63489 100644 --- a/cmd/internal/execution/flags.go +++ b/cmd/internal/execution/flags.go @@ -7,9 +7,16 @@ import ( cliruntime "github.com/MontFerret/cli/v2/pkg/runtime" ) +// AddRuntimeSelectionFlags registers runtime selection without execution policies. +func AddRuntimeSelectionFlags(cmd *cobra.Command) { + cmd.Flags().StringP(config.ExecRuntime, "r", cliruntime.DefaultRuntime, "Ferret runtime (builtin, wire, or legacy HTTP URL)") + cmd.Flags().String(config.ExecRuntimeEndpoint, "", "Wire endpoint: tcp://127.0.0.1: (trusted local development)") + cmd.Flags().Duration(config.ExecRuntimeConnectTimeout, cliruntime.DefaultConnectTimeout, "Wire connection and handshake timeout (must be positive)") +} + // AddRuntimeFlags keeps the execution-policy surface identical across run, debug, and repl. func AddRuntimeFlags(cmd *cobra.Command) { - cmd.Flags().StringP(config.ExecRuntime, "r", cliruntime.DefaultRuntime, "Ferret runtime type (\"builtin\"|$url)") + AddRuntimeSelectionFlags(cmd) cmd.Flags().String(config.ExecProxy, "x", "Proxy server address") cmd.Flags().String(config.ExecUserAgent, "a", "User agent header") cmd.Flags().StringP(config.ExecBrowserAddress, "d", cliruntime.DefaultBrowser, "Browser debugger address") diff --git a/cmd/internal/execution/runtime_options.go b/cmd/internal/execution/runtime_options.go index 17540df..e4d2c21 100644 --- a/cmd/internal/execution/runtime_options.go +++ b/cmd/internal/execution/runtime_options.go @@ -23,5 +23,5 @@ func OptionsFromCommand(cmd *cobra.Command, store *config.Store) (cliruntime.Opt } opts.FSPolicy = fsPolicy - return opts, nil + return opts, cliruntime.ValidateOptions(opts) } diff --git a/cmd/internal/execution/wire_test.go b/cmd/internal/execution/wire_test.go new file mode 100644 index 0000000..dc1b489 --- /dev/null +++ b/cmd/internal/execution/wire_test.go @@ -0,0 +1,140 @@ +package execution_test + +import ( + "strings" + "testing" + "time" + + "github.com/spf13/cobra" + + "github.com/MontFerret/cli/v2/cmd/internal/execution" + "github.com/MontFerret/cli/v2/cmd/internal/testutil" + "github.com/MontFerret/cli/v2/pkg/config" +) + +func TestWireExplicitConfiguration(t *testing.T) { + for _, extra := range [][]string{ + {}, {"--browser-open=false"}, {"--browser-headless=false"}, {"--browser-cookies=false"}, + {"--browser-address=http://127.0.0.1:9222"}, {"--proxy="}, {"--user-agent="}, + {"--policy-fs-read-only=false"}, {"--policy-http-follow-redirects=true"}, + } { + t.Run(strings.Join(extra, " "), func(t *testing.T) { + store := testutil.NewStore(t) + cmd := &cobra.Command{} + execution.AddRuntimeFlags(cmd) + args := append([]string{"--runtime=wire", "--runtime-endpoint=tcp://127.0.0.1:123"}, extra...) + if err := cmd.ParseFlags(args); err != nil { + t.Fatal(err) + } + + store.BindFlags(cmd) + opts, err := execution.OptionsFromCommand(cmd, store) + if (err != nil) != (len(extra) != 0) { + t.Fatalf("explicit options %v: %v", extra, err) + } + + if len(extra) == 0 && opts.ConnectTimeout != 5*time.Second { + t.Fatalf("default timeout = %v", opts.ConnectTimeout) + } + }) + } +} + +func TestWireConfigPrecedenceAndDefaults(t *testing.T) { + store := testutil.NewStore(t) + if err := store.Set(config.ExecRuntime, "wire"); err != nil { + t.Fatal(err) + } + + if err := store.Set(config.ExecRuntimeEndpoint, "tcp://127.0.0.1:111"); err != nil { + t.Fatal(err) + } + + if err := store.Set(config.ExecRuntimeConnectTimeout, "9s"); err != nil { + t.Fatal(err) + } + + store, err := config.NewStore("ferret", "test") + if err != nil { + t.Fatal(err) + } + + t.Setenv("FERRET_RUNTIME_ENDPOINT", "tcp://127.0.0.1:222") + t.Setenv("FERRET_RUNTIME_CONNECT_TIMEOUT", "7s") + cmd := &cobra.Command{} + execution.AddRuntimeFlags(cmd) + store.BindFlags(cmd) + opts, err := execution.OptionsFromCommand(cmd, store) + if err != nil || opts.Endpoint != "tcp://127.0.0.1:222" || opts.ConnectTimeout != 7*time.Second { + t.Fatalf("environment precedence: %+v error=%v", opts, err) + } + + cmd = &cobra.Command{} + execution.AddRuntimeFlags(cmd) + if err := cmd.ParseFlags([]string{"--runtime-endpoint=tcp://127.0.0.1:333", "--runtime-connect-timeout=3s"}); err != nil { + t.Fatal(err) + } + + store.BindFlags(cmd) + opts, err = execution.OptionsFromCommand(cmd, store) + if err != nil || opts.Endpoint != "tcp://127.0.0.1:333" || opts.ConnectTimeout != 3*time.Second { + t.Fatalf("flag precedence: %+v error=%v", opts, err) + } +} + +func TestWireConfigurationRejections(t *testing.T) { + for _, args := range [][]string{ + {"--runtime=wire"}, + {"--runtime=builtin", "--runtime-endpoint=tcp://127.0.0.1:1"}, + {"--runtime=https://worker.example", "--runtime-connect-timeout=5s"}, + {"--runtime=wire", "--runtime-endpoint=tcp://127.0.0.1:1", "--runtime-connect-timeout=0s"}, + {"--runtime=wire", "--runtime-endpoint=tcp://127.0.0.1:1", "--runtime-connect-timeout=-1s"}, + } { + t.Run(strings.Join(args, " "), func(t *testing.T) { + store := testutil.NewStore(t) + cmd := &cobra.Command{} + execution.AddRuntimeFlags(cmd) + if err := cmd.ParseFlags(args); err != nil { + t.Fatal(err) + } + + store.BindFlags(cmd) + if _, err := execution.OptionsFromCommand(cmd, store); err == nil { + t.Fatal("invalid configuration accepted") + } + }) + } +} + +func TestWireRejectsConfiguredPolicyAndBrowserDefaults(t *testing.T) { + for _, test := range []struct{ key, value string }{ + {config.ExecWithBrowser, "false"}, + {config.ExecBrowserAddress, "http://127.0.0.1:9222"}, + {config.PolicyFSReadOnly, "false"}, + {config.PolicyHTTPFollowRedirects, "true"}, + } { + for _, environment := range []bool{false, true} { + t.Run(test.key+"/"+map[bool]string{false: "config", true: "environment"}[environment], func(t *testing.T) { + store := testutil.NewStore(t) + if environment { + t.Setenv("FERRET_"+strings.ToUpper(strings.ReplaceAll(test.key, "-", "_")), test.value) + } else { + if err := store.Set(test.key, test.value); err != nil { + t.Fatal(err) + } + } + + cmd := &cobra.Command{} + execution.AddRuntimeFlags(cmd) + if err := cmd.ParseFlags([]string{"--runtime=wire", "--runtime-endpoint=tcp://127.0.0.1:1"}); err != nil { + t.Fatal(err) + } + + store.BindFlags(cmd) + if _, err := execution.OptionsFromCommand(cmd, store); err == nil { + t.Fatal("explicit builtin configuration accepted") + } + }) + } + } +} diff --git a/cmd/internal/repl/repl.go b/cmd/internal/repl/repl.go index 04ee102..6d55d64 100644 --- a/cmd/internal/repl/repl.go +++ b/cmd/internal/repl/repl.go @@ -44,7 +44,7 @@ func New(store *config.Store) *cobra.Command { defer cleanup() - return clirepl.Start(cmd.Context(), rtOpts, params) + return clirepl.StartWithIO(cmd.Context(), rtOpts, params, cmd.InOrStdin(), cmd.OutOrStdout(), cmd.ErrOrStderr()) }, } diff --git a/cmd/internal/repl/wire_test.go b/cmd/internal/repl/wire_test.go new file mode 100644 index 0000000..1f25ba4 --- /dev/null +++ b/cmd/internal/repl/wire_test.go @@ -0,0 +1,48 @@ +package repl + +import ( + "bytes" + "context" + "strings" + "testing" + "time" + + "github.com/MontFerret/cli/v2/cmd/internal/testutil" + "github.com/MontFerret/cli/v2/internal/testutil/wirehost" + "github.com/MontFerret/cli/v2/pkg/config" +) + +func TestWireReplReusesRuntimeAfterSyntaxError(t *testing.T) { + host := wirehost.New(t) + store := testutil.NewStore(t) + cmd := New(store) + cmd.SilenceErrors = true + cmd.SilenceUsage = true + cmd.SetArgs([]string{"--runtime", "wire", "--runtime-endpoint", host.Endpoint}) + cmd.SetIn(strings.NewReader("RETURN DEMO::IDENTITY(41)\nRETURN )\nRETURN DEMO::IDENTITY(42)\nexit\n")) + var stdout, stderr bytes.Buffer + cmd.SetOut(&stdout) + cmd.SetErr(&stderr) + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + + if err := cmd.ExecuteContext(config.With(ctx, store)); err != nil { + t.Fatal(err) + } + + for _, want := range []string{"Wire runtime: " + host.Endpoint + "; version: test-core-version", "41\n", "42\n"} { + if !strings.Contains(stdout.String(), want) { + t.Errorf("stdout missing %q: %q", want, stdout.String()) + } + } + + for _, want := range []string{"anonymous:1:8", "1 | RETURN )", " | ^"} { + if !strings.Contains(stderr.String(), want) { + t.Errorf("stderr missing %q: %q", want, stderr.String()) + } + } + + if host.Connections() != 1 { + t.Fatalf("connections=%d, want 1", host.Connections()) + } +} diff --git a/cmd/internal/run/array_diagnostics_test.go b/cmd/internal/run/array_diagnostics_test.go new file mode 100644 index 0000000..bde3fbb --- /dev/null +++ b/cmd/internal/run/array_diagnostics_test.go @@ -0,0 +1,40 @@ +package run + +import ( + "strings" + "testing" + + "github.com/MontFerret/cli/v2/internal/testutil/wirehost" +) + +func TestArrayDiagnosticsBuiltinAndWire(t *testing.T) { + host := wirehost.New(t) + for _, mode := range []string{"builtin", "wire"} { + for _, test := range []struct { + query string + want []string + }{ + {"let arr = [1, 2, 3", []string{"SyntaxError: Unclosed array literal\n", ":1:19\n", "1 | let arr = [1, 2, 3\n | ^ expected ']'\n", "Hint: Add ']' to close the array.\n"}}, + {"return [\n \"é🙂\"", []string{"SyntaxError: Unclosed array literal\n", ":2:11\n", "2 | \"é🙂\"\n | ^ expected ']'\n", "Hint: Add ']' to close the array.\n"}}, + {"return [1, 2 +", []string{"SyntaxError: Expected right-hand expression after '+'\n", "missing expression"}}, + } { + t.Run(mode+"/"+test.query, func(t *testing.T) { + args := []string{"run", "--runtime", mode, "--eval", test.query} + if mode == "wire" { + args = append(args, "--runtime-endpoint", host.Endpoint) + } + + stdout, stderr, err := runWireCommand(t, args) + if err == nil || stdout != "" || strings.Count(stderr, "SyntaxError:") != 1 { + t.Fatalf("stdout=%q stderr=%q error=%v", stdout, stderr, err) + } + + for _, want := range test.want { + if !strings.Contains(stderr, want) { + t.Errorf("stderr missing %q: %q", want, stderr) + } + } + }) + } + } +} diff --git a/cmd/internal/run/failing_close_reader_test.go b/cmd/internal/run/failing_close_reader_test.go new file mode 100644 index 0000000..6d79f5e --- /dev/null +++ b/cmd/internal/run/failing_close_reader_test.go @@ -0,0 +1,12 @@ +package run + +import "io" + +type failingCloseReader struct { + io.Reader + err error +} + +func (r *failingCloseReader) Close() error { + return r.err +} diff --git a/cmd/internal/run/result_test.go b/cmd/internal/run/result_test.go new file mode 100644 index 0000000..e183103 --- /dev/null +++ b/cmd/internal/run/result_test.go @@ -0,0 +1,70 @@ +package run + +import ( + "bytes" + "errors" + "io" + "strings" + "testing" + "testing/iotest" + + "github.com/spf13/cobra" +) + +func TestRunOutputWithError(t *testing.T) { + for _, test := range []struct { + output io.ReadCloser + want string + }{ + {nil, ""}, {io.NopCloser(strings.NewReader("")), ""}, {io.NopCloser(strings.NewReader("available")), "available"}, + } { + cmd := &cobra.Command{} + var stdout, stderr bytes.Buffer + cmd.SetOut(&stdout) + cmd.SetErr(&stderr) + failure := errors.New("execution or cleanup failed") + err := writeResult(cmd, test.output, failure) + if !errors.Is(err, failure) || !strings.Contains(stderr.String(), failure.Error()) { + t.Fatalf("output=%q stderr=%q error=%v", stdout.String(), stderr.String(), err) + } + + if stdout.String() != test.want { + t.Fatalf("output changed: %q", stdout.String()) + } + } +} + +func TestRunSuccessfulEmptyOutput(t *testing.T) { + cmd := &cobra.Command{} + var stdout, stderr bytes.Buffer + cmd.SetOut(&stdout) + cmd.SetErr(&stderr) + if err := writeResult(cmd, io.NopCloser(strings.NewReader("")), nil); err != nil || stdout.Len() != 0 || stderr.Len() != 0 { + t.Fatalf("empty result: stdout=%q stderr=%q error=%v", stdout.String(), stderr.String(), err) + } +} + +func TestRunJoinsExecutionCopyAndReaderCloseErrors(t *testing.T) { + executionErr := errors.New("execution failed") + copyErr := errors.New("reading output failed") + closeErr := errors.New("closing output failed") + output := &failingCloseReader{ + Reader: io.MultiReader(strings.NewReader("available"), iotest.ErrReader(copyErr)), + err: closeErr, + } + cmd := &cobra.Command{} + var stdout, stderr bytes.Buffer + cmd.SetOut(&stdout) + cmd.SetErr(&stderr) + + err := writeResult(cmd, output, executionErr) + for _, want := range []error{executionErr, copyErr, closeErr} { + if !errors.Is(err, want) || !strings.Contains(stderr.String(), want.Error()) { + t.Fatalf("failure lost: %v; stderr=%q", err, stderr.String()) + } + } + + if stdout.String() != "available" { + t.Fatalf("available output lost: %q", stdout.String()) + } +} diff --git a/cmd/internal/run/run.go b/cmd/internal/run/run.go index 4a0a920..a312545 100644 --- a/cmd/internal/run/run.go +++ b/cmd/internal/run/run.go @@ -1,9 +1,9 @@ package run import ( + "errors" "fmt" "io" - "os" "github.com/spf13/cobra" @@ -93,14 +93,18 @@ func execute(cmd *cobra.Command, rtOpts cliruntime.Options, brOpts browser.Optio out, err := clirun.Execute(cmd.Context(), rtOpts, params, input) - if err != nil { - diagnostics.PrintError(err) - return err - } + return writeResult(cmd, out, err) +} - defer out.Close() +func writeResult(cmd *cobra.Command, out io.ReadCloser, err error) error { + if out != nil { + _, copyErr := io.Copy(cmd.OutOrStdout(), out) + err = errors.Join(err, copyErr, out.Close()) + } - _, err = io.Copy(os.Stdout, out) + if err != nil { + diagnostics.PrintErrorTo(cmd.ErrOrStderr(), err) + } return err } diff --git a/cmd/internal/run/wire_test.go b/cmd/internal/run/wire_test.go new file mode 100644 index 0000000..fe10501 --- /dev/null +++ b/cmd/internal/run/wire_test.go @@ -0,0 +1,128 @@ +package run + +import ( + "bytes" + "context" + "errors" + "fmt" + "os" + "path/filepath" + "testing" + "time" + + "github.com/spf13/cobra" + + "github.com/MontFerret/api" + portable "github.com/MontFerret/api/diagnostics" + "github.com/MontFerret/wire/client" + "github.com/MontFerret/wire/pkg/failure" + + "github.com/MontFerret/cli/v2/cmd/internal/testutil" + "github.com/MontFerret/cli/v2/internal/testutil/wirehost" + "github.com/MontFerret/cli/v2/pkg/config" + cliruntime "github.com/MontFerret/cli/v2/pkg/runtime" +) + +func TestWireRunCommands(t *testing.T) { + host := wirehost.New(t) + for _, mode := range []string{"inline", "file", "exec", "stdin"} { + t.Run(mode, func(t *testing.T) { + query := "RETURN DEMO::IDENTITY([@n, @b, @arr, @obj, @str, @nothing])" + args := []string{"run", "--runtime", "wire", "--runtime-endpoint", host.Endpoint, + "--param", "n=42", "--param", "b=true", "--param", "arr=[1,2]", "--param", `obj={"value":false}`, "--param", `str="123"`, "--param", "nothing=null"} + + switch mode { + case "exec": + args[0] = "exec" + args = append(args, "--eval", query) + case "inline": + args = append(args, "--eval", query) + case "file": + path := filepath.Join(t.TempDir(), "query.fql") + testutil.WriteQuery(t, path, query) + args = append(args, path) + } + + invoke := func() { + stdout, stderr, err := runWireCommand(t, args) + if err != nil || stderr != "" || stdout != `[42,true,[1,2],{"value":false},"123",null]` { + t.Fatalf("stdout=%q stderr=%q error=%v", stdout, stderr, err) + } + } + + if mode == "stdin" { + testutil.WithStdinBytes(t, []byte(query), invoke) + } else { + invoke() + } + }) + } +} + +func TestWireRunRenderedDiagnostics(t *testing.T) { + host := wirehost.New(t) + for _, test := range []struct { + query, rendered string + span api.Span + }{ + {"let arr = []\n\nreturn", "SyntaxError: Expected expression after 'return'\n --> %s:3:7\n |\n2 | \n3 | return\n | ^ missing return value\nHint: Did you forget to provide a value to return?\n", api.Span{Start: 20, End: 20}}, + {"RETURN )", "SyntaxError: Expected expression after 'RETURN'\n --> %s:1:8\n |\n1 | RETURN )\n | ^ missing return value\nHint: Did you forget to provide a value to return?\n", api.Span{Start: 7, End: 8}}, + {"let arr = [1, 2, 3", "SyntaxError: Unclosed array literal\n --> %s:1:19\n |\n1 | let arr = [1, 2, 3\n | ^ expected ']'\nHint: Add ']' to close the array.\n", api.Span{Start: 18, End: 18}}, + {"return [1, 2 +", "SyntaxError: Expected right-hand expression after '+'\n --> %s:1:15\n |\n1 | return [1, 2 +\n | ^ missing expression\nHint: Provide an expression after the operator.\n", api.Span{Start: 14, End: 14}}, + } { + t.Run(test.query, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "invalid.fql") + testutil.WriteQuery(t, path, test.query) + stdout, stderr, err := runWireCommand(t, []string{"run", "--runtime", "wire", "--runtime-endpoint", host.Endpoint, path}) + if err == nil || stdout != "" { + t.Fatalf("stdout=%q error=%v", stdout, err) + } + + if want := fmt.Sprintf(test.rendered, path); stderr != want { + t.Errorf("stderr = %q, want %q", stderr, want) + } + + var values portable.Diagnostics + var rpc *client.Error + var terminal *failure.Failure + if errors.As(err, &rpc) { + values = rpc.Diagnostics + } else if errors.As(err, &terminal) { + values = terminal.Diagnostics + } + + if len(values) != 1 || len(values[0].Annotations) != 1 || values[0].Annotations[0].Range.Span != test.span || !values[0].Annotations[0].Primary { + t.Fatalf("incorrect host diagnostic range: %+v", values) + } + }) + } +} + +func TestWireArtifactRejectedBeforeConnecting(t *testing.T) { + host := wirehost.New(t) + path := filepath.Join(t.TempDir(), "query.fqlc") + if err := os.WriteFile(path, []byte("FBC2"), 0600); err != nil { + t.Fatal(err) + } + + _, _, err := runWireCommand(t, []string{"run", "--runtime", "wire", "--runtime-endpoint", host.Endpoint, path}) + if !errors.Is(err, cliruntime.ErrArtifactRequiresBuiltinRuntime) || host.Connections() != 0 { + t.Fatalf("error=%v connections=%d", err, host.Connections()) + } +} + +func runWireCommand(t *testing.T, args []string) (string, string, error) { + t.Helper() + store := testutil.NewStore(t) + root := &cobra.Command{SilenceErrors: true, SilenceUsage: true} + root.AddCommand(New(store)) + var stdout, stderr bytes.Buffer + root.SetOut(&stdout) + root.SetErr(&stderr) + root.SetArgs(args) + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + err := root.ExecuteContext(config.With(ctx, store)) + + return stdout.String(), stderr.String(), err +} diff --git a/cmd/internal/testutil/testutil.go b/cmd/internal/testutil/testutil.go index 63ba94f..11e1c68 100644 --- a/cmd/internal/testutil/testutil.go +++ b/cmd/internal/testutil/testutil.go @@ -5,10 +5,13 @@ import ( "io" "os" "path/filepath" + "strings" "testing" + "github.com/mitchellh/go-homedir" "github.com/spf13/cobra" + "github.com/MontFerret/cli/v2/pkg/config" "github.com/MontFerret/ferret/v2/pkg/compiler" ) @@ -24,6 +27,25 @@ func WriteQuery(t *testing.T, path, content string) { } } +// NewStore isolates command configuration from the user's filesystem and environment. +func NewStore(t *testing.T) *config.Store { + t.Helper() + t.Setenv("HOME", t.TempDir()) + for _, key := range config.Flags { + t.Setenv("FERRET_"+strings.ToUpper(strings.ReplaceAll(key, "-", "_")), "") + } + + homedir.Reset() + t.Cleanup(homedir.Reset) + + store, err := config.NewStore("ferret", "test") + if err != nil { + t.Fatal(err) + } + + return store +} + func CaptureStdout(t *testing.T, fn func() error) (string, error) { t.Helper() diff --git a/cmd/internal/version/failing_writer_test.go b/cmd/internal/version/failing_writer_test.go new file mode 100644 index 0000000..ac9060a --- /dev/null +++ b/cmd/internal/version/failing_writer_test.go @@ -0,0 +1,9 @@ +package version + +type failingWriter struct { + err error +} + +func (w failingWriter) Write([]byte) (int, error) { + return 0, w.err +} diff --git a/cmd/internal/version/version.go b/cmd/internal/version/version.go index b2d6294..4a2f35f 100644 --- a/cmd/internal/version/version.go +++ b/cmd/internal/version/version.go @@ -1,10 +1,12 @@ package version import ( + "errors" "fmt" "github.com/spf13/cobra" + "github.com/MontFerret/cli/v2/cmd/internal/execution" "github.com/MontFerret/cli/v2/pkg/config" "github.com/MontFerret/cli/v2/pkg/runtime" ) @@ -12,7 +14,7 @@ import ( func New(store *config.Store) *cobra.Command { cmd := &cobra.Command{ Use: "version", - Short: "Show the CLI version information", + Short: "Show the CLI and selected runtime version information", Args: cobra.MaximumNArgs(0), PreRun: func(cmd *cobra.Command, _ []string) { store.BindFlags(cmd) @@ -22,29 +24,32 @@ func New(store *config.Store) *cobra.Command { }, } - cmd.Flags().StringP(config.ExecRuntime, "r", runtime.DefaultRuntime, "Ferret runtime type (\"builtin\"|$url)") + execution.AddRuntimeSelectionFlags(cmd) return cmd } -func runVersion(cmd *cobra.Command, store *config.Store) error { - rt, err := runtime.New(store.GetRuntimeOptions()) - +func runVersion(cmd *cobra.Command, store *config.Store) (err error) { + resources, err := runtime.New(cmd.Context(), store.GetRuntimeOptions()) if err != nil { return err } - defer rt.Close() + defer func() { err = errors.Join(err, resources.Close()) }() - ver, err := rt.Version(cmd.Context()) + var version string + if resources.Legacy != nil { + version, err = resources.Legacy.Version(cmd.Context()) + } else { + metadata, versionErr := resources.Runtime.Version(cmd.Context()) + version, err = metadata.String(), versionErr + } if err != nil { return err } - fmt.Println("Version:") - fmt.Printf(" Self: %s\n", store.AppVersion()) - fmt.Printf(" Runtime: %s\n", ver) + _, err = fmt.Fprintf(cmd.OutOrStdout(), "Version:\n Self: %s\n Runtime: %s\n", store.AppVersion(), version) - return nil + return err } diff --git a/cmd/internal/version/wire_test.go b/cmd/internal/version/wire_test.go new file mode 100644 index 0000000..f065fc2 --- /dev/null +++ b/cmd/internal/version/wire_test.go @@ -0,0 +1,257 @@ +package version + +import ( + "bytes" + "context" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/MontFerret/api" + "github.com/MontFerret/cli/v2/cmd/internal/testutil" + "github.com/MontFerret/cli/v2/internal/testutil/wirehost" + "github.com/MontFerret/cli/v2/pkg/config" + "github.com/MontFerret/cli/v2/pkg/runtime" + "github.com/MontFerret/wire/client" + "github.com/MontFerret/wire/pkg/failure" +) + +func TestWireVersionCommand(t *testing.T) { + for _, value := range []api.Version{"v2.0.0-alpha.57", " runtime+opaque \n", "版本-α", "runtime\x00\xff", ""} { + t.Run(string(value), func(t *testing.T) { + host := wirehost.NewWithVersion(t, value) + store := testutil.NewStore(t) + stdout, stderr, err := executeVersion(t.Context(), store, []string{"--runtime", "wire", "--runtime-endpoint", host.Endpoint}) + want := fmt.Sprintf("Version:\n Self: test\n Runtime: %s\n", value) + if err != nil || stdout != want || stderr != "" || host.Connections() != 1 { + t.Fatalf("stdout=%q stderr=%q error=%v connections=%d; want %q", stdout, stderr, err, host.Connections(), want) + } + + waitForVersionCleanup(t, host) + + if _, err := host.Runtime.Run(t.Context(), api.NewSource("survivor.fql", "RETURN DEMO::IDENTITY(42)")); err != nil { + t.Fatalf("version command closed host engine: %v", err) + } + }) + } +} + +func TestVersionBuiltinAndLegacyOutput(t *testing.T) { + for _, mode := range []string{"builtin", "legacy"} { + t.Run(mode, func(t *testing.T) { + selected, expected := "builtin", runtime.EmbeddedVersion() + if mode == "legacy" { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/info" { + t.Errorf("version request path=%q", r.URL.Path) + } + + _, _ = fmt.Fprint(w, `{"version":{"ferret":"legacy-version"}}`) + })) + defer server.Close() + selected, expected = server.URL, "legacy-version" + } + + stdout, stderr, err := executeVersion(t.Context(), testutil.NewStore(t), []string{"--runtime", selected}) + want := fmt.Sprintf("Version:\n Self: test\n Runtime: %s\n", expected) + if err != nil || stdout != want || stderr != "" { + t.Fatalf("stdout=%q stderr=%q error=%v; want %q", stdout, stderr, err, want) + } + }) + } +} + +func TestWireVersionConfigurationPrecedence(t *testing.T) { + host := wirehost.NewWithVersion(t, "selected-host") + store := testutil.NewStore(t) + for key, value := range map[string]string{config.ExecRuntime: "wire", config.ExecRuntimeEndpoint: "tcp://127.0.0.1:1", config.ExecRuntimeConnectTimeout: "0s"} { + if err := store.Set(key, value); err != nil { + t.Fatal(err) + } + } + + store, err := config.NewStore("ferret", "test") + if err != nil { + t.Fatal(err) + } + + t.Setenv("FERRET_RUNTIME_ENDPOINT", host.Endpoint) + t.Setenv("FERRET_RUNTIME_CONNECT_TIMEOUT", "5s") + stdout, _, err := executeVersion(t.Context(), store, nil) + if err != nil || !strings.Contains(stdout, "Runtime: selected-host\n") { + t.Fatalf("environment precedence: stdout=%q error=%v", stdout, err) + } + + store, err = config.NewStore("ferret", "test") + if err != nil { + t.Fatal(err) + } + + t.Setenv("FERRET_RUNTIME_ENDPOINT", "tcp://127.0.0.1:1") + t.Setenv("FERRET_RUNTIME_CONNECT_TIMEOUT", "0s") + stdout, _, err = executeVersion(t.Context(), store, []string{"--runtime-endpoint", host.Endpoint, "--runtime-connect-timeout", "5s"}) + if err != nil || !strings.Contains(stdout, "Runtime: selected-host\n") || host.Connections() != 2 { + t.Fatalf("flag precedence: stdout=%q error=%v connections=%d", stdout, err, host.Connections()) + } +} + +func TestWireVersionValidatesBeforeConnecting(t *testing.T) { + host := wirehost.New(t) + for _, args := range [][]string{ + {"--runtime", "wire"}, + {"--runtime", "wire", "--runtime-endpoint", "tcp://localhost:1"}, + {"--runtime", "wire", "--runtime-endpoint", host.Endpoint, "--runtime-connect-timeout=0s"}, + {"--runtime", "builtin", "--runtime-endpoint", host.Endpoint}, + {"--runtime", "http://worker.example", "--runtime-connect-timeout=5s"}, + } { + stdout, _, err := executeVersion(t.Context(), testutil.NewStore(t), args) + if err == nil || stdout != "" || host.Connections() != 0 { + t.Fatalf("accepted options %v: stdout=%q error=%v connections=%d", args, stdout, err, host.Connections()) + } + } + + for _, setting := range []struct{ key, value string }{ + {config.ExecWithBrowser, "false"}, {config.ExecBrowserAddress, runtime.DefaultBrowser}, + {config.PolicyFSReadOnly, "false"}, {config.PolicyHTTPFollowRedirects, "true"}, + } { + for _, environment := range []bool{false, true} { + t.Run(setting.key+"/"+fmt.Sprint(environment), func(t *testing.T) { + store := testutil.NewStore(t) + if environment { + t.Setenv("FERRET_"+strings.ToUpper(strings.ReplaceAll(setting.key, "-", "_")), setting.value) + } else if err := store.Set(setting.key, setting.value); err != nil { + t.Fatal(err) + } + + stdout, _, err := executeVersion(t.Context(), store, []string{"--runtime", "wire", "--runtime-endpoint", host.Endpoint}) + if err == nil || stdout != "" || host.Connections() != 0 { + t.Fatalf("accepted explicit configuration: stdout=%q error=%v connections=%d", stdout, err, host.Connections()) + } + }) + } + } +} + +func TestWireVersionHandshakeFailure(t *testing.T) { + host := wirehost.NewWithVersionFunc(t, func(context.Context) (api.Version, error) { + return "", errors.New("private hosted metadata error") + }) + stdout, stderr, err := executeVersion(t.Context(), testutil.NewStore(t), []string{"--runtime", "wire", "--runtime-endpoint", host.Endpoint}) + var wireErr *client.Error + if !errors.As(err, &wireErr) || wireErr.Category != failure.CategoryInternalRuntime || stdout != "" || stderr != "" || strings.Contains(err.Error(), "private") { + t.Fatalf("failed metadata: stdout=%q stderr=%q error=%v", stdout, stderr, err) + } + + waitForVersionCleanup(t, host) + + if !strings.Contains(err.Error(), host.Endpoint) || host.Connections() != 1 { + t.Fatalf("missing endpoint context or unexpected retry: error=%v connections=%d", err, host.Connections()) + } +} + +func TestWireVersionOutputFailureReleasesClient(t *testing.T) { + host := wirehost.New(t) + cmd := New(testutil.NewStore(t)) + cmd.SilenceErrors, cmd.SilenceUsage = true, true + cmd.SetArgs([]string{"--runtime", "wire", "--runtime-endpoint", host.Endpoint}) + writeErr := errors.New("output is closed") + cmd.SetOut(failingWriter{err: writeErr}) + if err := cmd.ExecuteContext(t.Context()); !errors.Is(err, writeErr) { + t.Fatalf("lost output error: %v", err) + } + + waitForVersionCleanup(t, host) + + if _, err := host.Runtime.Run(t.Context(), api.NewSource("survivor.fql", "RETURN DEMO::IDENTITY(42)")); err != nil { + t.Fatalf("output failure closed hosted runtime: %v", err) + } +} + +func waitForVersionCleanup(t *testing.T, host *wirehost.Host) { + t.Helper() + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + + if err := host.WaitForClientsClosed(ctx); err != nil { + t.Fatalf("client transport was not released: %v", err) + } +} + +func executeVersion(ctx context.Context, store *config.Store, args []string) (string, string, error) { + cmd := New(store) + cmd.SilenceErrors, cmd.SilenceUsage = true, true + cmd.SetArgs(args) + var stdout, stderr bytes.Buffer + cmd.SetOut(&stdout) + cmd.SetErr(&stderr) + ctx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + err := cmd.ExecuteContext(ctx) + + return stdout.String(), stderr.String(), err +} + +func TestWireVersionCommandCancellation(t *testing.T) { + started, finished := make(chan struct{}), make(chan struct{}) + host := wirehost.NewWithVersionFunc(t, func(ctx context.Context) (api.Version, error) { + close(started) + defer close(finished) + <-ctx.Done() + + return "", ctx.Err() + }) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + store := testutil.NewStore(t) + done := make(chan error, 1) + go func() { + stdout, _, err := executeVersion(ctx, store, []string{"--runtime", "wire", "--runtime-endpoint", host.Endpoint}) + if stdout != "" { + err = errors.Join(err, fmt.Errorf("canceled metadata produced output: %q", stdout)) + } + + done <- err + }() + select { + case <-started: + case <-time.After(10 * time.Second): + t.Fatal("host did not receive metadata request") + } + + cancel() + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("lost command cancellation: %v", err) + } + case <-time.After(10 * time.Second): + t.Fatal("version command did not stop") + } + + select { + case <-finished: + case <-time.After(10 * time.Second): + t.Fatal("hosted metadata request did not settle") + } + + waitForVersionCleanup(t, host) +} + +func TestVersionRegistersOnlyRuntimeSelectionFlags(t *testing.T) { + cmd := New(testutil.NewStore(t)) + for _, name := range []string{config.ExecRuntime, config.ExecRuntimeEndpoint, config.ExecRuntimeConnectTimeout} { + if cmd.Flags().Lookup(name) == nil { + t.Errorf("missing selection flag --%s", name) + } + } + + for _, name := range []string{config.ExecWithBrowser, config.ExecBrowserAddress, config.ExecProxy, config.ExecUserAgent, config.PolicyFSRoot, config.PolicyHTTPFollowRedirects} { + if cmd.Flags().Lookup(name) != nil { + t.Errorf("unexpected execution flag --%s", name) + } + } +} diff --git a/examples/wire-host/main.go b/examples/wire-host/main.go new file mode 100644 index 0000000..92315ec --- /dev/null +++ b/examples/wire-host/main.go @@ -0,0 +1,96 @@ +// wire-host exposes an application-configured runtime for trusted local development. +package main + +import ( + "context" + "errors" + "fmt" + "net" + "os" + "os/signal" + "syscall" + "time" + + "github.com/MontFerret/api" + "github.com/MontFerret/cli/v2/internal/buildinfo" + "github.com/MontFerret/ferret/v2" + "github.com/MontFerret/ferret/v2/pkg/runtime" + "github.com/MontFerret/ferret/v2/uapi" + "github.com/MontFerret/wire/server" +) + +func main() { + if err := run(); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} + +func run() (err error) { + version, err := buildinfo.FerretVersion() + if err != nil { + return err + } + + root, err := os.Getwd() + if err != nil { + return err + } + + engine, err := ferret.New(ferret.WithFSRoot(root), ferret.WithFunctionsRegistrar(func(ns runtime.Namespace) { + ns.Namespace("DEMO").Function().A1().Add("IDENTITY", func(_ context.Context, value runtime.Value) (runtime.Value, error) { + return value, nil + }) + })) + if err != nil { + return err + } + + defer func() { err = errors.Join(err, engine.Close()) }() + hosted := uapi.Wrap(engine, api.Version(version)) + defer func() { err = errors.Join(err, hosted.Close()) }() + wire, err := server.NewServer(hosted) + if err != nil { + return err + } + + defer func() { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + err = errors.Join(err, wire.Shutdown(ctx)) + }() + + listener, err := net.Listen("tcp4", "127.0.0.1:0") + if err != nil { + return err + } + + defer func() { + if closeErr := listener.Close(); closeErr != nil && !errors.Is(closeErr, net.ErrClosed) { + err = errors.Join(err, closeErr) + } + }() + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + fmt.Printf("tcp://%s\n", listener.Addr()) + served := make(chan error, 1) + go func() { served <- wire.Serve(context.Background(), listener) }() + + select { + case err := <-served: + return err + case <-ctx.Done(): + } + + shutdown, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + err = wire.Shutdown(shutdown) + + select { + case serveErr := <-served: + return errors.Join(err, serveErr) + case <-shutdown.Done(): + return errors.Join(err, shutdown.Err()) + } +} diff --git a/go.mod b/go.mod index 7aa530d..495ea51 100644 --- a/go.mod +++ b/go.mod @@ -4,6 +4,7 @@ go 1.26.5 require ( github.com/Masterminds/semver/v3 v3.5.0 + github.com/MontFerret/api v1.0.0-alpha.20 github.com/MontFerret/barn v1.4.0 github.com/MontFerret/contrib/modules/ai/llm v1.0.0-rc.6 github.com/MontFerret/contrib/modules/archive v1.0.0-rc.5 @@ -24,6 +25,7 @@ require ( github.com/MontFerret/contrib/modules/yaml v1.0.0-rc.16 github.com/MontFerret/ferret/v2 v2.0.0-alpha.57 github.com/MontFerret/specs v1.12.0 + github.com/MontFerret/wire v1.0.0-alpha.2 github.com/antlr4-go/antlr/v4 v4.13.1 github.com/chzyer/readline v1.5.1 github.com/go-waitfor/waitfor v1.1.0 @@ -40,13 +42,13 @@ require ( github.com/spf13/viper v1.21.0 golang.org/x/mod v0.41.0 golang.org/x/sync v0.23.0 + google.golang.org/grpc v1.84.0 ) require ( github.com/BurntSushi/toml v1.6.0 // indirect github.com/JohannesKaufmann/dom v0.3.1 // indirect github.com/JohannesKaufmann/html-to-markdown/v2 v2.5.2 // indirect - github.com/MontFerret/api v1.0.0-alpha.20 // indirect github.com/MontFerret/contrib/pkg/common v0.2.0 // indirect github.com/MontFerret/cssx v0.2.0 // indirect github.com/PuerkitoBio/goquery v1.13.0 // indirect @@ -108,6 +110,8 @@ require ( golang.org/x/sys v0.48.0 // indirect golang.org/x/text v0.42.0 // indirect golang.org/x/tools v0.50.0 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260810153831-ec0a7760b754 // indirect + google.golang.org/protobuf v1.36.12 // indirect gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect gopkg.in/yaml.v2 v2.2.8 // indirect modernc.org/libc v1.75.7 // indirect diff --git a/go.sum b/go.sum index 6018610..bfd5de5 100644 --- a/go.sum +++ b/go.sum @@ -54,6 +54,8 @@ github.com/MontFerret/ferret/v2 v2.0.0-alpha.57 h1:E2KcHVQBLnY0A0QkIyewN5DGj9TJr github.com/MontFerret/ferret/v2 v2.0.0-alpha.57/go.mod h1:/qxBUibHjztUbhG6XH7ErPffepGBdm0nW3JWrqBVVWA= github.com/MontFerret/specs v1.12.0 h1:1HaElGLN+tyZIsrb3WVdulxbW5W0LXEdOH0k6F1sFzk= github.com/MontFerret/specs v1.12.0/go.mod h1:ENGyEPFjrWVPWOUAsBMc1acUsqpswieTrAJfhcm576E= +github.com/MontFerret/wire v1.0.0-alpha.2 h1:8ttKxRV5CA3IMpC6GEw8rBGoA3j13n6khhApl4LOK38= +github.com/MontFerret/wire v1.0.0-alpha.2/go.mod h1:/64SY8tt4W0pd6Pn3JPeVDY/ITs6DsIe8ZHOQ2w4SI4= github.com/PuerkitoBio/goquery v1.13.0 h1:mqHbjD7Jmnul4DTR24LKTjo1uUmHUh072kteGV+xpFM= github.com/PuerkitoBio/goquery v1.13.0/go.mod h1:Hip5mdBL8K2wEGKJdr27sRaNwIdDajmCwB/ExUPwW+g= github.com/andybalholm/cascadia v1.3.5 h1:RLjq12WJy58dN6eCIQrz0bAGZkztHWsEPFxP53Y7Ms8= @@ -111,6 +113,8 @@ github.com/golang-jwt/jwt/v5 v5.3.1/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArs github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc= github.com/golang/groupcache v0.0.0-20241129210726-2c02b8208cf8 h1:f+oWsMOmNPc8JmEHVZIycC7hBoQxHH9pNKQORJNozsQ= github.com/golang/groupcache v0.0.0-20241129210726-2c02b8208cf8/go.mod h1:wcDNUvekVysuuOpQKo3191zZyTpiI6se1N1ULghS0sw= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= @@ -331,6 +335,14 @@ golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d/go.mod h1:aiJjzUbINMkxb golang.org/x/tools v0.50.0 h1:c2ifzfcuY7L90lZ2aKd8S4K2NpASF08SZx9ZuJkHmSU= golang.org/x/tools v0.50.0/go.mod h1:7ulVMw3831Mwi5EZD6RomGyffr4VFjuNYXf2BbCEAV0= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= +gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260810153831-ec0a7760b754 h1:k5CJw9e5ONCcA/u0webKt092npXuY+KeGh3Q8NAVf0g= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260810153831-ec0a7760b754/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.84.0 h1:soMyaPJ8pAak5PIQ0DGBUir0XRo2fRoMqhNWMLlLxO0= +google.golang.org/grpc v1.84.0/go.mod h1:ljCht0DrxQrXBDRTZp52Qxh3Ffk8CdYm2sj4O2QN2C0= +google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc= +google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/natefinch/lumberjack.v2 v2.2.1 h1:bBRl1b0OH9s/DuPhuXpNl+VtCaJXFZ5/uEFST95x9zc= gopkg.in/natefinch/lumberjack.v2 v2.2.1/go.mod h1:YD8tP3GAjkrDg1eZH7EGmyESg/lsYskCTPBJVb9jqSc= diff --git a/internal/diagnostics/diagnostics.go b/internal/diagnostics/diagnostics.go new file mode 100644 index 0000000..8232e7b --- /dev/null +++ b/internal/diagnostics/diagnostics.go @@ -0,0 +1,198 @@ +// Package diagnostics presents native and portable execution failures. +package diagnostics + +import ( + "encoding/json" + "fmt" + "io" + "strings" + + "github.com/MontFerret/api" + portable "github.com/MontFerret/api/diagnostics" + native "github.com/MontFerret/ferret/v2/pkg/diagnostics" + "github.com/MontFerret/ferret/v2/pkg/source" + "github.com/MontFerret/wire/client" + "github.com/MontFerret/wire/pkg/failure" +) + +// Format renders structured diagnostics throughout joined and wrapped errors. +// Portable projections take precedence over their duplicate native causes; +// unrelated failures, including cleanup failures, remain visible. +func Format(err error) string { + var out strings.Builder + projected := make(map[string]bool) + visit(err, func(current error) { + for _, diagnostic := range portableAt(current) { + projected[diagnosticKey(diagnostic)] = true + } + }) + printed := make(map[string]bool) + render(&out, err, projected, printed) + + return strings.TrimRight(out.String(), "\n") +} + +// Print writes a formatted execution failure to the selected diagnostic stream. +func Print(out io.Writer, err error) { + if err != nil { + fmt.Fprintln(out, Format(err)) + } +} + +func portableAt(err error) portable.Diagnostics { + switch value := err.(type) { + case portable.Diagnostics: + return value + case *client.Error: + if value != nil { + return value.Diagnostics + } + case *failure.Failure: + if value != nil { + return value.Diagnostics + } + } + + return nil +} + +func visit(err error, fn func(error)) { + if err == nil { + return + } + + fn(err) + + switch value := err.(type) { + case interface{ Unwrap() []error }: + for _, child := range value.Unwrap() { + visit(child, fn) + } + case interface{ Unwrap() error }: + visit(value.Unwrap(), fn) + } +} + +func render(out io.Writer, err error, projected, printed map[string]bool) { + if err == nil { + return + } + + if values := portableAt(err); len(values) > 0 { + for _, value := range values { + key := diagnosticKey(value) + if !printed[key] { + printed[key] = true + formatPortable(out, presentation(value)) + } + } + + return + } + + if value, ok := err.(*native.Diagnostic); ok && value != nil { + if projected[diagnosticKey(portableDiagnostic(value))] { + render(out, value.Cause, projected, printed) + } else if separateCause(value.Cause, projected) { + projectedValue := *value + projectedValue.Cause = nil + native.FormatDiagnostic(out, &projectedValue, 0) + render(out, value.Cause, projected, printed) + } else { + native.FormatDiagnostic(out, value, 0) + } + + return + } + + switch value := err.(type) { + case interface{ Unwrap() []error }: + for _, child := range value.Unwrap() { + render(out, child, projected, printed) + } + case interface{ Unwrap() error }: + structured := false + visit(value.Unwrap(), func(child error) { + if len(portableAt(child)) > 0 { + structured = true + } + + if _, ok := child.(*native.Diagnostic); ok { + structured = true + } + }) + + if structured { + render(out, value.Unwrap(), projected, printed) + } else { + fmt.Fprintln(out, native.Format(err)) + } + default: + fmt.Fprintln(out, native.Format(err)) + } +} + +// Native's formatter follows one diagnostic cause. Split structured branches +// when that would hide joined failures or repeat an already portable projection. +func separateCause(err error, projected map[string]bool) bool { + separate := false + visit(err, func(current error) { + if _, ok := current.(interface{ Unwrap() []error }); ok || len(portableAt(current)) > 0 { + separate = true + } + + if value, ok := current.(*native.Diagnostic); ok && value != nil && projected[diagnosticKey(portableDiagnostic(value))] { + separate = true + } + }) + + return separate +} + +func presentation(value portable.Diagnostic) *native.Diagnostic { + result := &native.Diagnostic{ + Kind: native.Kind(value.Kind), Message: value.Message, + Source: source.New(value.Source.Name, value.Source.Content), + Hint: value.Hint, Note: value.Note, + } + + for _, annotation := range value.Annotations { + result.Spans = append(result.Spans, native.ErrorSpan{ + Span: source.Span{Start: annotation.Range.Span.Start, End: annotation.Range.Span.End}, + Label: annotation.Message, Main: annotation.Primary, + }) + } + + return result +} + +func portableDiagnostic(value *native.Diagnostic) portable.Diagnostic { + result := portable.Diagnostic{ + Kind: portable.Kind(value.Kind), Message: value.Message, + Source: api.NewSource(value.Source.Name(), value.Source.Content()), + Hint: value.Hint, Note: value.Note, + } + + for _, span := range value.Spans { + rangeAt := value.Source.RangeAt(span.Span) + result.Annotations = append(result.Annotations, portable.Annotation{ + Range: api.Range{ + Location: api.Location{SourceName: rangeAt.SourceName, Position: api.Position{Line: rangeAt.Line, Column: rangeAt.Column}}, + Span: api.Span{Start: span.Span.Start, End: span.Span.End}, + }, + Message: span.Label, Primary: span.Main, + }) + } + + return result +} + +func diagnosticKey(value portable.Diagnostic) string { + if len(value.Annotations) == 0 { + value.Annotations = nil + } + + encoded, _ := json.Marshal(value) + + return string(encoded) +} diff --git a/internal/diagnostics/diagnostics_test.go b/internal/diagnostics/diagnostics_test.go new file mode 100644 index 0000000..4253cf3 --- /dev/null +++ b/internal/diagnostics/diagnostics_test.go @@ -0,0 +1,112 @@ +package diagnostics + +import ( + "errors" + "fmt" + "strings" + "testing" + + "github.com/MontFerret/api" + portable "github.com/MontFerret/api/diagnostics" + native "github.com/MontFerret/ferret/v2/pkg/diagnostics" + "github.com/MontFerret/wire/client" + "github.com/MontFerret/wire/pkg/failure" +) + +func TestPortablePresentation(t *testing.T) { + diagnostic := portable.Diagnostic{ + Source: api.NewSource("query.fql", "RETURN"), Kind: "SyntaxError", Message: "Missing value", + Annotations: []portable.Annotation{{Range: api.Range{Location: api.Location{SourceName: "query.fql", Position: api.Position{Line: 1, Column: 7}}, Span: api.Span{Start: 6, End: 6}}, Message: "missing value", Primary: true}}, + Hint: "Provide a value", Note: "Produced by host", + } + values := portable.Diagnostics{diagnostic} + cleanup := errors.New("close physical connection failed") + for _, err := range []error{values, &client.Error{Category: failure.CategoryCompilation, Diagnostics: values}, &failure.Failure{Category: failure.CategoryCompilation, Diagnostics: values}} { + text := Format(errors.Join(fmt.Errorf("execute: %w", err), cleanup)) + want := "SyntaxError: Missing value\n --> query.fql:1:7\n |\n1 | RETURN\n | ^ missing value\nNote: Produced by host\nHint: Provide a value\nclose physical connection failed" + if text != want { + t.Fatalf("presentation = %q, want %q", text, want) + } + } + + text := Format(errors.Join(values, presentation(diagnostic))) + if strings.Count(text, "SyntaxError: Missing value") != 1 { + t.Fatalf("duplicate projection:\n%s", text) + } + + if !Recoverable(values) || Recoverable(errors.Join(values, cleanup)) || Recoverable(&client.Error{Category: failure.CategoryConnectionNotFound}) { + t.Fatal("incorrect recoverable classification") + } +} + +func TestPortableAnnotationsAndInvalidCoordinates(t *testing.T) { + value := portable.Diagnostic{ + Source: api.NewSource("query.fql", "RETURN )"), Kind: "CustomHostKind", Message: "Unexpected token", + Annotations: []portable.Annotation{ + {Range: api.Range{Span: api.Span{Start: 0, End: 6}}, Message: "statement"}, + {Range: api.Range{Span: api.Span{Start: 7, End: 8}}, Message: "offending token", Primary: true}, + }, + } + text := Format(portable.Diagnostics{value}) + for _, want := range []string{"CustomHostKind", "^^^^^^ statement", " ^ offending token", "query.fql:1:8"} { + if !strings.Contains(text, want) { + t.Errorf("missing %q:\n%s", want, text) + } + } + + value.Annotations[1].Range.Span.End = 9 + adapted := presentation(value) + if adapted.Cause != nil || adapted.Spans[1].Span.End != 9 || adapted.Spans[1].Main != true { + t.Fatal("coordinates or primary marker changed") + } + + if strings.Contains(Format(portable.Diagnostics{value}), "offending token") { + t.Fatal("invalid coordinate was clamped") + } +} + +func TestNativePresentationUnchanged(t *testing.T) { + err := &native.Diagnostic{Kind: native.UnexpectedError, Message: "native failure"} + if Format(err) != strings.TrimSpace(native.Format(err)) { + t.Fatal("native rendering changed") + } +} + +func TestJoinedNativeCausesRetainCleanupWithoutDuplicateProjection(t *testing.T) { + outer := &native.Diagnostic{Kind: native.UnexpectedError, Message: "outer failure"} + inner := &native.Diagnostic{Kind: native.UnexpectedError, Message: "inner failure"} + cleanup := errors.New("cleanup failed") + outer.Cause = errors.Join(inner, cleanup) + projections := portable.Diagnostics{portableDiagnostic(outer), portableDiagnostic(inner)} + + for _, err := range []error{outer, errors.Join(projections, outer)} { + text := Format(err) + for _, want := range []string{"outer failure", "inner failure", "cleanup failed"} { + if strings.Count(text, want) != 1 { + t.Fatalf("lost or duplicated %q:\n%s", want, text) + } + } + } +} + +func TestPortableMultiplePrimaryAnnotations(t *testing.T) { + value := portable.Diagnostic{ + Source: api.NewSource("query.fql", "RETURN )"), Kind: "CustomHostKind", Message: "Two related ranges", + Annotations: []portable.Annotation{ + {Range: api.Range{Span: api.Span{Start: 0, End: 6}}, Message: "first primary", Primary: true}, + {Range: api.Range{Span: api.Span{Start: 7, End: 8}}, Message: "second primary", Primary: true}, + }, + Note: "Both ranges matter", Hint: "Correct both", + } + text := Format(portable.Diagnostics{value}) + want := "CustomHostKind: Two related ranges\n --> query.fql:1:1\n |\n1 | RETURN )\n | ^^^^^^ first primary\n --> query.fql:1:8\n |\n1 | RETURN )\n | ^ second primary\nNote: Both ranges matter\nHint: Correct both" + if text != want { + t.Fatalf("presentation = %q, want %q", text, want) + } + + for _, span := range presentation(value).Spans { + if !span.Main { + t.Fatal("primary marker changed") + } + } +} diff --git a/internal/diagnostics/portable_format.go b/internal/diagnostics/portable_format.go new file mode 100644 index 0000000..d26ceae --- /dev/null +++ b/internal/diagnostics/portable_format.go @@ -0,0 +1,50 @@ +package diagnostics + +import ( + "fmt" + "io" + "sort" + + native "github.com/MontFerret/ferret/v2/pkg/diagnostics" +) + +func formatPortable(out io.Writer, diagnostic *native.Diagnostic) { + primaries := 0 + for _, span := range diagnostic.Spans { + if span.Main { + primaries++ + } + } + + if primaries <= 1 { + native.FormatDiagnostic(out, diagnostic, 0) + + return + } + + // Native's full formatter selects a single primary span. Keep all portable + // primary markers and use its span renderer for collections with several. + header := *diagnostic + header.Spans = nil + header.Note = "" + header.Hint = "" + native.FormatDiagnostic(out, &header, 0) + spans := append([]native.ErrorSpan(nil), diagnostic.Spans...) + sort.SliceStable(spans, func(i, j int) bool { return spans[i].Span.Start < spans[j].Span.Start }) + renderer := native.SpanRenderer{CaretChar: '^'} + for _, primary := range []bool{false, true} { + for _, span := range spans { + if span.Main == primary { + renderer.Render(out, diagnostic.Source, span.Span, span.Label) + } + } + } + + if diagnostic.Note != "" { + fmt.Fprintf(out, "Note: %s\n", diagnostic.Note) + } + + if diagnostic.Hint != "" { + fmt.Fprintf(out, "Hint: %s\n", diagnostic.Hint) + } +} diff --git a/internal/diagnostics/recoverable.go b/internal/diagnostics/recoverable.go new file mode 100644 index 0000000..5fa96a8 --- /dev/null +++ b/internal/diagnostics/recoverable.go @@ -0,0 +1,52 @@ +package diagnostics + +import ( + portable "github.com/MontFerret/api/diagnostics" + native "github.com/MontFerret/ferret/v2/pkg/diagnostics" + "github.com/MontFerret/wire/client" + "github.com/MontFerret/wire/pkg/failure" +) + +// Recoverable reports query failures that leave the runtime usable. Joined +// transport or cleanup failures are terminal even when a query diagnostic is present. +func Recoverable(err error) bool { + if err == nil { + return false + } + + switch value := err.(type) { + case portable.Diagnostics: + return len(value) > 0 + case *native.Diagnostic: + return value != nil + case *client.Error: + return value != nil && queryCategory(value.Category) + case *failure.Failure: + return value != nil && queryCategory(value.Category) + case interface{ Unwrap() []error }: + children := value.Unwrap() + found := false + + for _, child := range children { + if child == nil { + continue + } + + if !Recoverable(child) { + return false + } + + found = true + } + + return found + case interface{ Unwrap() error }: + return Recoverable(value.Unwrap()) + default: + return false + } +} + +func queryCategory(category failure.Category) bool { + return category == failure.CategoryCompilation || category == failure.CategoryExecution +} diff --git a/internal/testutil/wirehost/host.go b/internal/testutil/wirehost/host.go new file mode 100644 index 0000000..0d92432 --- /dev/null +++ b/internal/testutil/wirehost/host.go @@ -0,0 +1,178 @@ +// Package wirehost provides a configured native runtime over real loopback TCP +// for CLI tests and benchmarks. It uses only public Ferret and Wire APIs. +package wirehost + +import ( + "context" + "errors" + "net" + "sync" + "testing" + "time" + + "github.com/MontFerret/api" + "github.com/MontFerret/ferret/v2" + native "github.com/MontFerret/ferret/v2/pkg/runtime" + "github.com/MontFerret/ferret/v2/uapi" + "github.com/MontFerret/wire/server" +) + +// Host owns its configured engine, server, and listener independently of clients. +type Host struct { + Endpoint string + Runtime api.Runtime + WaitStarted chan struct{} + WaitFinished chan struct{} + engine *ferret.Engine + server *server.Server + listener *listener + served chan error + once sync.Once + err error +} + +// New starts a bounded, test-owned host with DEMO::IDENTITY and DEMO::WAIT. +func New(t testing.TB) *Host { + return NewWithVersion(t, "test-core-version") +} + +// NewWithVersion hosts an opaque runtime version, including an empty value. +func NewWithVersion(t testing.TB, version api.Version) *Host { + return newHost(t, version, nil) +} + +// NewWithVersionFunc controls hosted metadata retrieval for handshake failure tests. +func NewWithVersionFunc(t testing.TB, version func(context.Context) (api.Version, error)) *Host { + return newHost(t, "", version) +} + +func newHost(t testing.TB, version api.Version, retrieve func(context.Context) (api.Version, error)) *Host { + t.Helper() + h := &Host{WaitStarted: make(chan struct{}), WaitFinished: make(chan struct{})} + var started, finished sync.Once + var err error + h.engine, err = ferret.New(ferret.WithFSRoot(t.TempDir()), ferret.WithFunctionsRegistrar(func(ns native.Namespace) { + demo := ns.Namespace("DEMO") + demo.Function().A1().Add("IDENTITY", func(_ context.Context, value native.Value) (native.Value, error) { + return value, nil + }) + demo.Function().A0().Add("WAIT", func(ctx context.Context) (native.Value, error) { + started.Do(func() { close(h.WaitStarted) }) + defer finished.Do(func() { close(h.WaitFinished) }) + <-ctx.Done() + + return nil, ctx.Err() + }) + })) + if err != nil { + t.Fatal(err) + } + + t.Cleanup(func() { + if err := h.Close(); err != nil { + t.Errorf("close Wire fixture: %v", err) + } + }) + h.Runtime = uapi.Wrap(h.engine, version) + if retrieve != nil { + h.Runtime = &versionRuntime{Runtime: h.Runtime, retrieve: retrieve} + } + + h.server, err = server.NewServer(h.Runtime) + if err != nil { + t.Fatal(err) + } + + ln, err := net.Listen("tcp4", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + + h.listener = &listener{Listener: ln} + h.Endpoint = "tcp://" + ln.Addr().String() + h.served = make(chan error, 1) + go func() { h.served <- h.server.Serve(context.Background(), h.listener) }() + + return h +} + +// Connections counts physical connections accepted by the fixture. +func (h *Host) Connections() int64 { + return h.listener.accepted.Load() +} + +// WaitForClientsClosed observes transport release without polling or closing clients. +func (h *Host) WaitForClientsClosed(ctx context.Context) error { + for { + h.listener.mu.Lock() + if len(h.listener.connections) == 0 { + h.listener.mu.Unlock() + + return nil + } + + if h.listener.changed == nil { + h.listener.changed = make(chan struct{}) + } + + changed := h.listener.changed + h.listener.mu.Unlock() + + select { + case <-changed: + case <-ctx.Done(): + return ctx.Err() + } + } +} + +// StopTransport simulates a broken transport without closing the host engine. +func (h *Host) StopTransport() { + h.listener.mu.Lock() + connections := make([]net.Conn, 0, len(h.listener.connections)) + for connection := range h.listener.connections { + connections = append(connections, connection) + } + h.listener.mu.Unlock() + + for _, connection := range connections { + _ = connection.Close() + } +} + +// Close stops the server before releasing the host-owned engine. +func (h *Host) Close() error { + h.once.Do(func() { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + if h.server != nil { + h.err = errors.Join(h.err, h.server.Shutdown(ctx)) + } + + if h.listener != nil { + if err := h.listener.Close(); err != nil && !errors.Is(err, net.ErrClosed) { + h.err = errors.Join(h.err, err) + } + } + + if h.served != nil { + select { + case err := <-h.served: + h.err = errors.Join(h.err, err) + case <-ctx.Done(): + h.err = errors.Join(h.err, ctx.Err()) + } + } + + if h.Runtime != nil { + h.err = errors.Join(h.err, h.Runtime.Close()) + } + + if h.engine != nil { + h.err = errors.Join(h.err, h.engine.Close()) + } + }) + + return h.err +} diff --git a/internal/testutil/wirehost/listener.go b/internal/testutil/wirehost/listener.go new file mode 100644 index 0000000..a25d8fa --- /dev/null +++ b/internal/testutil/wirehost/listener.go @@ -0,0 +1,34 @@ +package wirehost + +import ( + "net" + "sync" + "sync/atomic" +) + +type listener struct { + net.Listener + accepted atomic.Int64 + mu sync.Mutex + connections map[net.Conn]struct{} + changed chan struct{} +} + +func (l *listener) Accept() (net.Conn, error) { + connection, err := l.Listener.Accept() + if err != nil { + return nil, err + } + + tracked := &trackedConnection{Conn: connection, listener: l} + l.mu.Lock() + if l.connections == nil { + l.connections = make(map[net.Conn]struct{}) + } + + l.connections[tracked] = struct{}{} + l.mu.Unlock() + l.accepted.Add(1) + + return tracked, nil +} diff --git a/internal/testutil/wirehost/tracked_connection.go b/internal/testutil/wirehost/tracked_connection.go new file mode 100644 index 0000000..29f17d5 --- /dev/null +++ b/internal/testutil/wirehost/tracked_connection.go @@ -0,0 +1,23 @@ +package wirehost + +import "net" + +type trackedConnection struct { + net.Conn + listener *listener +} + +func (c *trackedConnection) Close() error { + err := c.Conn.Close() + c.listener.mu.Lock() + delete(c.listener.connections, c) + + if c.listener.changed != nil { + close(c.listener.changed) + c.listener.changed = nil + } + + c.listener.mu.Unlock() + + return err +} diff --git a/internal/testutil/wirehost/version_runtime.go b/internal/testutil/wirehost/version_runtime.go new file mode 100644 index 0000000..5772221 --- /dev/null +++ b/internal/testutil/wirehost/version_runtime.go @@ -0,0 +1,17 @@ +package wirehost + +import ( + "context" + + "github.com/MontFerret/api" +) + +// versionRuntime changes only metadata retrieval; execution and ownership stay native. +type versionRuntime struct { + api.Runtime + retrieve func(context.Context) (api.Version, error) +} + +func (r *versionRuntime) Version(ctx context.Context) (api.Version, error) { + return r.retrieve(ctx) +} diff --git a/pkg/config/flags.go b/pkg/config/flags.go index 1e9862f..648b2b2 100644 --- a/pkg/config/flags.go +++ b/pkg/config/flags.go @@ -7,13 +7,15 @@ const ( LoggerOutput = "log-output" LoggerFile = "log-file" - ExecRuntime = "runtime" - ExecKeepCookies = "browser-cookies" - ExecWithBrowser = "browser-open" - ExecBrowserAddress = "browser-address" - ExecWithBrowserHeadless = "browser-headless" - ExecProxy = "proxy" - ExecUserAgent = "user-agent" + ExecRuntime = "runtime" + ExecRuntimeEndpoint = "runtime-endpoint" + ExecRuntimeConnectTimeout = "runtime-connect-timeout" + ExecKeepCookies = "browser-cookies" + ExecWithBrowser = "browser-open" + ExecBrowserAddress = "browser-address" + ExecWithBrowserHeadless = "browser-headless" + ExecProxy = "proxy" + ExecUserAgent = "user-agent" PolicyFSRoot = "policy-fs-root" PolicyFSReadOnly = "policy-fs-read-only" @@ -47,6 +49,8 @@ var Flags = []string{ LoggerOutput, LoggerFile, ExecRuntime, + ExecRuntimeEndpoint, + ExecRuntimeConnectTimeout, ExecKeepCookies, ExecBrowserAddress, ExecWithBrowser, diff --git a/pkg/config/runtime_metadata_test.go b/pkg/config/runtime_metadata_test.go new file mode 100644 index 0000000..59db974 --- /dev/null +++ b/pkg/config/runtime_metadata_test.go @@ -0,0 +1,44 @@ +package config + +import ( + "strings" + "testing" + + "github.com/mitchellh/go-homedir" + "github.com/spf13/cobra" + + "github.com/MontFerret/cli/v2/pkg/runtime" +) + +func TestRuntimeMetadataOptionsDetectUnboundConfiguration(t *testing.T) { + for _, test := range []struct{ key, value string }{ + {PolicyFSReadOnly, "false"}, {PolicyHTTPFollowRedirects, "true"}, {ExecWithBrowser, "false"}, + } { + t.Run(test.key, func(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + homedir.Reset() + t.Cleanup(homedir.Reset) + store, err := NewStore("ferret", "test") + if err != nil { + t.Fatal(err) + } + + cmd := &cobra.Command{} + cmd.Flags().String(ExecRuntime, runtime.DefaultRuntime, "") + cmd.Flags().String(ExecRuntimeEndpoint, "", "") + if err := cmd.ParseFlags([]string{"--runtime=wire", "--runtime-endpoint=tcp://127.0.0.1:1"}); err != nil { + t.Fatal(err) + } + + store.BindFlags(cmd) + if err := runtime.ValidateOptions(store.GetRuntimeOptions()); err != nil { + t.Fatalf("untouched defaults rejected: %v", err) + } + + t.Setenv("FERRET_"+strings.ToUpper(strings.ReplaceAll(test.key, "-", "_")), test.value) + if err := runtime.ValidateOptions(store.GetRuntimeOptions()); err == nil { + t.Fatal("explicit environmental default was ignored without a registered flag") + } + }) + } +} diff --git a/pkg/config/store.go b/pkg/config/store.go index 2e53e89..e6e9afb 100644 --- a/pkg/config/store.go +++ b/pkg/config/store.go @@ -49,10 +49,9 @@ func NewStore(appName, version string) (*Store, error) { // binds to an environment variable STING_NUMBER. This helps // avoid conflicts. v.SetEnvPrefix(envPrefix) + v.SetEnvKeyReplacer(strings.NewReplacer("-", "_")) - // Bind to environment variables - // Works great for simple config names, but needs help for names - // like --favorite-color which we fix in the bindFlags function + // Read settings even when the selected command does not register their flags. v.AutomaticEnv() return &Store{appName, version, envPrefix, v}, nil @@ -106,6 +105,29 @@ func (s *Store) GetRuntimeOptions() runtime.Options { opts.Type = s.v.GetString(ExecRuntime) } + opts.EndpointSet = s.v.IsSet(ExecRuntimeEndpoint) + if opts.EndpointSet { + opts.Endpoint = s.v.GetString(ExecRuntimeEndpoint) + } + + opts.ConnectTimeoutSet = s.v.IsSet(ExecRuntimeConnectTimeout) + if opts.ConnectTimeoutSet { + opts.ConnectTimeout = s.v.GetDuration(ExecRuntimeConnectTimeout) + } + + for _, key := range []string{ExecBrowserAddress, ExecKeepCookies, ExecWithBrowser, ExecWithBrowserHeadless, ExecProxy, ExecUserAgent} { + if s.v.IsSet(key) { + opts.BrowserOptionsSet = true + } + } + + for _, key := range Flags { + if strings.HasPrefix(key, "policy-") && s.v.IsSet(key) { + opts.BuiltinPolicyOptionsSet = true + break + } + } + if s.v.IsSet(ExecBrowserAddress) { opts.BrowserAddress = s.v.GetString(ExecBrowserAddress) } diff --git a/pkg/repl/compatibility_test.go b/pkg/repl/compatibility_test.go new file mode 100644 index 0000000..0cb8c3b --- /dev/null +++ b/pkg/repl/compatibility_test.go @@ -0,0 +1,59 @@ +package repl + +import ( + "bytes" + "context" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/MontFerret/cli/v2/pkg/runtime" +) + +func TestBuiltinReplBannerAndSyntaxRecovery(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + var stdout, stderr bytes.Buffer + err := StartWithIO(ctx, runtime.NewDefaultOptions(), nil, strings.NewReader("RETURN 41\nRETURN )\nRETURN 42\nexit\n"), &stdout, &stderr) + want := "Welcome to Ferret REPL " + runtime.EmbeddedVersion() + "\nPlease use `exit` or `Ctrl-D` to exit this program.\n41\n42\n" + if err != nil || stdout.String() != want || !strings.Contains(stderr.String(), "anonymous:1:8") { + t.Fatalf("stdout=%q stderr=%q error=%v", stdout.String(), stderr.String(), err) + } +} + +func TestLegacyHTTPReplBannerAndRequests(t *testing.T) { + records := make(chan string, 3) + worker := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + records <- r.Method + " " + r.URL.Path + if r.URL.Path == "/info" { + _, _ = io.WriteString(w, `{"version":{"ferret":"legacy-host"}}`) + + return + } + + _, _ = io.Copy(io.Discard, r.Body) + _, _ = io.WriteString(w, "42") + })) + defer worker.Close() + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + var stdout, stderr bytes.Buffer + err := StartWithIO(ctx, runtime.Options{Type: worker.URL}, nil, strings.NewReader("RETURN 42\nRETURN 42\nexit\n"), &stdout, &stderr) + var requests []string + for range 3 { + select { + case request := <-records: + requests = append(requests, request) + case <-ctx.Done(): + t.Fatal("legacy requests did not complete") + } + } + + want := "Welcome to Ferret REPL legacy-host\nPlease use `exit` or `Ctrl-D` to exit this program.\n42\n42\n" + if err != nil || stdout.String() != want || stderr.Len() != 0 || strings.Join(requests, ",") != "GET /info,POST /,POST /" { + t.Fatalf("stdout=%q stderr=%q requests=%v error=%v", stdout.String(), stderr.String(), requests, err) + } +} diff --git a/pkg/repl/lifecycle_test.go b/pkg/repl/lifecycle_test.go new file mode 100644 index 0000000..0e5b0d5 --- /dev/null +++ b/pkg/repl/lifecycle_test.go @@ -0,0 +1,118 @@ +package repl + +import ( + "bytes" + "context" + "errors" + "io" + goruntime "runtime" + "strings" + "testing" + "time" + + "github.com/MontFerret/cli/v2/internal/testutil/wirehost" + "github.com/MontFerret/cli/v2/pkg/runtime" +) + +func TestWireReplIdleCancellation(t *testing.T) { + testWireReplIdleStop(t, false) +} + +func TestWireReplIdleTransportLoss(t *testing.T) { + testWireReplIdleStop(t, true) +} + +func testWireReplIdleStop(t *testing.T, transportLoss bool) { + t.Helper() + host := wirehost.New(t) + reader, writer := io.Pipe() + defer reader.Close() + defer writer.Close() + reading := make(chan struct{}) + input := readFunc(func(data []byte) (int, error) { + close(reading) + + return reader.Read(data) + }) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + var stdout, stderr bytes.Buffer + done := make(chan error, 1) + go func() { + done <- StartWithIO(ctx, runtime.Options{Type: "wire", Endpoint: host.Endpoint}, nil, input, &stdout, &stderr) + }() + select { + case <-reading: + case <-time.After(10 * time.Second): + t.Fatal("REPL did not request input") + } + + if transportLoss { + host.StopTransport() + } else { + cancel() + } + + select { + case err := <-done: + if err == nil || (!transportLoss && !errors.Is(err, context.Canceled)) || (transportLoss && !strings.Contains(err.Error(), "check the runtime host")) { + t.Fatalf("idle stop: %v", err) + } + case <-time.After(10 * time.Second): + t.Fatal("idle REPL did not exit") + } +} + +func TestWireReplExecutionStops(t *testing.T) { + for _, transportLoss := range []bool{false, true} { + t.Run(map[bool]string{false: "cancel", true: "transport loss"}[transportLoss], func(t *testing.T) { + host := wirehost.New(t) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + var stdout, stderr bytes.Buffer + done := make(chan error, 1) + go func() { + done <- StartWithIO(ctx, runtime.Options{Type: "wire", Endpoint: host.Endpoint}, nil, strings.NewReader("RETURN DEMO::WAIT()\n"), &stdout, &stderr) + }() + select { + case <-host.WaitStarted: + case err := <-done: + t.Fatalf("REPL returned before query started: %v\nstdout=%q\nstderr=%q", err, stdout.String(), stderr.String()) + case <-time.After(10 * time.Second): + goroutines := make([]byte, 1<<20) + goroutines = goroutines[:goruntime.Stack(goroutines, true)] + cancel() + + // The REPL owns these writers until it returns. Do not inspect + // their buffers if cancellation fails to settle the goroutine. + select { + case err := <-done: + t.Fatalf("query did not start; REPL returned after cancellation: %v\nstdout=%q\nstderr=%q\ngoroutines before cancellation:\n%s", err, stdout.String(), stderr.String(), goroutines) + case <-time.After(10 * time.Second): + t.Fatalf("query did not start and REPL did not settle after cancellation\ngoroutines before cancellation:\n%s", goroutines) + } + } + + if transportLoss { + host.StopTransport() + } else { + cancel() + } + + select { + case err := <-done: + if err == nil || (!transportLoss && !errors.Is(err, context.Canceled)) || (transportLoss && !strings.Contains(err.Error(), "check the runtime host")) { + t.Fatalf("REPL stop error = %v", err) + } + case <-time.After(10 * time.Second): + t.Fatal("REPL did not settle") + } + + select { + case <-host.WaitFinished: + case <-time.After(10 * time.Second): + t.Fatal("host work leaked") + } + }) + } +} diff --git a/pkg/repl/metadata_test.go b/pkg/repl/metadata_test.go new file mode 100644 index 0000000..9fec25d --- /dev/null +++ b/pkg/repl/metadata_test.go @@ -0,0 +1,88 @@ +package repl + +import ( + "bytes" + "context" + "errors" + "fmt" + "strings" + "testing" + "time" + + "github.com/MontFerret/api" + "github.com/MontFerret/cli/v2/internal/testutil/wirehost" + "github.com/MontFerret/cli/v2/pkg/runtime" +) + +func TestWireReplHostedVersion(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + for _, version := range []api.Version{" runtime+opaque \n", "版本-α", ""} { + t.Run(string(version), func(t *testing.T) { + host := wirehost.NewWithVersion(t, version) + var stdout, stderr bytes.Buffer + err := StartWithIO(ctx, runtime.Options{Type: "wire", Endpoint: host.Endpoint}, nil, + strings.NewReader("RETURN DEMO::IDENTITY(41)\nRETURN DEMO::IDENTITY(42)\nexit\n"), &stdout, &stderr) + want := fmt.Sprintf("Welcome to Ferret REPL (Wire runtime: %s; version: %s)\nPlease use `exit` or `Ctrl-D` to exit this program.\n41\n42\n", host.Endpoint, version) + if err != nil || stdout.String() != want || stderr.Len() != 0 || host.Connections() != 1 { + t.Fatalf("stdout=%q stderr=%q error=%v connections=%d; want %q", stdout.String(), stderr.String(), err, host.Connections(), want) + } + }) + } +} + +func TestWireReplMetadataFailureDoesNotReadInput(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + host := wirehost.NewWithVersionFunc(t, func(context.Context) (api.Version, error) { + return "", errors.New("hosted metadata failed") + }) + var stdout, stderr bytes.Buffer + input := readFunc(func([]byte) (int, error) { + t.Error("requested input after failed metadata") + + return 0, errors.New("unexpected input") + }) + err := StartWithIO(ctx, runtime.Options{Type: "wire", Endpoint: host.Endpoint}, nil, input, &stdout, &stderr) + if err == nil || stdout.Len() != 0 || stderr.Len() != 0 || host.Connections() != 1 { + t.Fatalf("metadata failure: stdout=%q stderr=%q error=%v connections=%d", stdout.String(), stderr.String(), err, host.Connections()) + } +} + +func TestWireReplBannerWriteFailureReleasesClient(t *testing.T) { + for _, failAt := range []int{1, 2} { + t.Run(fmt.Sprint(failAt), func(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + host := wirehost.New(t) + writeErr := errors.New("banner output is closed") + writes := 0 + output := writeFunc(func(data []byte) (int, error) { + writes++ + if writes == failAt { + return 0, writeErr + } + + return len(data), nil + }) + input := readFunc(func([]byte) (int, error) { + t.Error("requested input after banner output failure") + + return 0, errors.New("unexpected input") + }) + var stderr bytes.Buffer + err := StartWithIO(ctx, runtime.Options{Type: "wire", Endpoint: host.Endpoint}, nil, input, output, &stderr) + if !errors.Is(err, writeErr) || writes != failAt || stderr.Len() != 0 { + t.Fatalf("banner failure: writes=%d stderr=%q error=%v", writes, stderr.String(), err) + } + + if err := host.WaitForClientsClosed(ctx); err != nil { + t.Fatalf("banner failure leaked client transport: %v", err) + } + + if _, err := host.Runtime.Run(ctx, api.NewSource("survivor.fql", "RETURN DEMO::IDENTITY(42)")); err != nil { + t.Fatalf("banner failure closed the host runtime: %v", err) + } + }) + } +} diff --git a/pkg/repl/read_func_test.go b/pkg/repl/read_func_test.go new file mode 100644 index 0000000..321ab29 --- /dev/null +++ b/pkg/repl/read_func_test.go @@ -0,0 +1,7 @@ +package repl + +type readFunc func([]byte) (int, error) + +func (f readFunc) Read(data []byte) (int, error) { + return f(data) +} diff --git a/pkg/repl/repl.go b/pkg/repl/repl.go index a389194..69f4cb1 100644 --- a/pkg/repl/repl.go +++ b/pkg/repl/repl.go @@ -2,72 +2,152 @@ package repl import ( "context" + "errors" "fmt" + "io" "os" "strings" + "sync" "github.com/chzyer/readline" "github.com/MontFerret/ferret/v2/pkg/source" + "github.com/MontFerret/cli/v2/internal/diagnostics" "github.com/MontFerret/cli/v2/pkg/runtime" ) +// Start opens an interactive source shell using the process's standard streams. func Start(ctx context.Context, opts runtime.Options, params map[string]interface{}) error { - rt, err := runtime.New(opts) + return StartWithIO(ctx, opts, params, os.Stdin, os.Stdout, os.Stderr) +} +// StartWithIO borrows command streams and owns its readline and runtime resources. +// Parent cancellation interrupts input and execution; active execution settles +// before the runtime and physical transport are released. +func StartWithIO(ctx context.Context, opts runtime.Options, params map[string]any, stdin io.Reader, stdout, stderr io.Writer) (err error) { + resources, err := runtime.New(ctx, opts) if err != nil { return err } - defer rt.Close() + defer func() { err = errors.Join(err, resources.Close()) }() - version, err := rt.Version(ctx) + var version string + if resources.Legacy != nil { + version, err = resources.Legacy.Version(ctx) + } else { + metadata, versionErr := resources.Runtime.Version(ctx) + version, err = metadata.String(), versionErr + } if err != nil { return err } - fmt.Printf("Welcome to Ferret REPL %s\n", version) - fmt.Println("Please use `exit` or `Ctrl-D` to exit this program.") - - rl, err := readline.NewEx(&readline.Config{ - Prompt: "> ", - InterruptPrompt: "^C", - EOFPrompt: "exit", - }) + if runtime.IsWireType(opts.Type) { + _, err = fmt.Fprintf(stdout, "Welcome to Ferret REPL (Wire runtime: %s; version: %s)\n", opts.Endpoint, version) + } else { + _, err = fmt.Fprintf(stdout, "Welcome to Ferret REPL %s\n", version) + } if err != nil { return err } - defer rl.Close() + if _, err := fmt.Fprintln(stdout, "Please use `exit` or `Ctrl-D` to exit this program."); err != nil { + return err + } - var commands []string - var multiline bool + input := readline.NewCancelableStdin(stdin) + defer input.Close() + + readlineConfig := &readline.Config{ + Prompt: "> ", InterruptPrompt: "^C", EOFPrompt: "exit", + Stdin: input, Stdout: stdout, Stderr: stderr, + } + + if stdin != os.Stdin || stdout != os.Stdout { + // Injected streams must not manipulate the process terminal or depend + // on whether the calling test or embedding application has a TTY. + readlineConfig.FuncIsTerminal = func() bool { return false } + readlineConfig.FuncMakeRaw = func() error { return nil } + readlineConfig.FuncExitRaw = func() error { return nil } + readlineConfig.FuncOnWidthChanged = func(func()) {} + } + + rl, err := readline.NewEx(readlineConfig) + if err != nil { + return err + } - ctx, cancel := context.WithCancel(context.Background()) + ctx, cancel := context.WithCancelCause(ctx) + defer cancel(nil) + + if resources.Disconnected != nil { + watched := make(chan struct{}) + go func() { + defer close(watched) + + select { + case <-ctx.Done(): + case <-resources.Disconnected: + cancel(fmt.Errorf("wire runtime %s disconnected; check the runtime host", opts.Endpoint)) + } + }() + defer func() { + cancel(nil) + <-watched + }() + } - exit := func() { - cancel() + var closeOnce sync.Once + var closeErr error + closeInput := func() { + closeOnce.Do(func() { + // Interrupt only input asynchronously. Close readline after Readline + // returns so its terminal goroutine has completed initialization. + closeErr = input.Close() + }) } + closed := make(chan struct{}) + stop := context.AfterFunc(ctx, func() { + closeInput() + close(closed) + }) + defer func() { + if !stop() { + <-closed + } + + closeInput() + err = errors.Join(err, closeErr, rl.Close()) + }() + + var commands []string + var multiline bool for { - line, err := rl.Readline() + line, readErr := rl.Readline() + if ctx.Err() != nil { + return context.Cause(ctx) + } + + if readErr != nil { + if errors.Is(readErr, io.EOF) || errors.Is(readErr, readline.ErrInterrupt) { + return nil + } - if err != nil { - break + return readErr } line = strings.TrimSpace(line) - if line == "" { continue } if strings.HasPrefix(line, "%") { line = line[1:] - multiline = !multiline } @@ -79,32 +159,37 @@ func Start(ctx context.Context, opts runtime.Options, params map[string]interfac commands = append(commands, line) query := strings.TrimSpace(strings.Join(commands, "\n")) commands = make([]string, 0, 10) - if query == "" { continue } if query == "exit" { - exit() + return nil + } - break + out, runErr := runtime.RunSource(ctx, resources, source.NewAnonymous(query), params) + if out != nil { + if outputErr := writeResult(stdout, out); outputErr != nil { + return errors.Join(runErr, outputErr) + } } - out, err := rt.Run(ctx, source.NewAnonymous(query), params) + if runErr != nil { + diagnostics.Print(stderr, runErr) - if err != nil { - fmt.Println("Failed to execute the query") - fmt.Println(err) - continue - } + if ctx.Err() != nil { + return errors.Join(context.Cause(ctx), runErr) + } - err = writeResult(os.Stdout, out) + if diagnostics.Recoverable(runErr) || resources.Legacy != nil { + continue + } - if err != nil { - fmt.Fprintln(os.Stderr, err) - break + if runtime.IsWireType(opts.Type) { + return fmt.Errorf("wire runtime %s is unavailable; check the runtime host: %w", opts.Endpoint, runErr) + } + + return runErr } } - - return nil } diff --git a/pkg/repl/write_func_test.go b/pkg/repl/write_func_test.go new file mode 100644 index 0000000..88380aa --- /dev/null +++ b/pkg/repl/write_func_test.go @@ -0,0 +1,7 @@ +package repl + +type writeFunc func([]byte) (int, error) + +func (f writeFunc) Write(data []byte) (int, error) { + return f(data) +} diff --git a/pkg/runtime/builtin.go b/pkg/runtime/builtin.go index 0660b64..ec05e33 100644 --- a/pkg/runtime/builtin.go +++ b/pkg/runtime/builtin.go @@ -13,7 +13,6 @@ import ( "github.com/MontFerret/ferret/v2/pkg/logging" ferretnet "github.com/MontFerret/ferret/v2/pkg/net" ferrethttp "github.com/MontFerret/ferret/v2/pkg/net/http" - "github.com/MontFerret/ferret/v2/pkg/source" ) var version = "unknown" @@ -28,10 +27,6 @@ type Builtin struct { network ferretnet.Network } -func NewBuiltin(opts Options) (Runtime, error) { - return newBuiltin(opts) -} - func newBuiltin(opts Options) (*Builtin, error) { fsRoot := "" if opts.FSPolicy != nil { @@ -80,8 +75,7 @@ func newBuiltin(opts Options) (*Builtin, error) { if len(opts.HTTPPolicy) > 0 { client, err := ferrethttp.New(opts.HTTPPolicy...) if err != nil { - _ = log.Close() - return nil, fmt.Errorf("initialize HTTP policy: %w", err) + return nil, errors.Join(fmt.Errorf("initialize HTTP policy: %w", err), log.Close()) } network, err = ferretnet.New(ferretnet.WithHTTPClient(client)) @@ -90,8 +84,7 @@ func newBuiltin(opts Options) (*Builtin, error) { closer.CloseIdleConnections() } - _ = log.Close() - return nil, fmt.Errorf("initialize network: %w", err) + return nil, errors.Join(fmt.Errorf("initialize network: %w", err), log.Close()) } engineOpts = append(engineOpts, ferret.WithNetwork(network)) @@ -104,8 +97,7 @@ func newBuiltin(opts Options) (*Builtin, error) { ferretnet.CloseIdleNetworkConnections(network) } - _ = log.Close() - return nil, fmt.Errorf("initialize engine: %w", err) + return nil, errors.Join(fmt.Errorf("initialize engine: %w", err), log.Close()) } return &Builtin{ @@ -116,20 +108,6 @@ func newBuiltin(opts Options) (*Builtin, error) { }, nil } -func (rt *Builtin) Version(_ context.Context) (string, error) { - return version, nil -} - -func (rt *Builtin) Run(ctx context.Context, query source.Source, params map[string]any) (io.ReadCloser, error) { - res, err := rt.engine.Run(ctx, query, ferret.WithSessionParams(params)) - - if err != nil { - return nil, err - } - - return io.NopCloser(bytes.NewBuffer(res.Content)), nil -} - func (rt *Builtin) RunArtifact(ctx context.Context, data []byte, params map[string]any) (io.ReadCloser, error) { plan, err := rt.engine.Load(data) diff --git a/pkg/runtime/builtin_benchmark_test.go b/pkg/runtime/builtin_benchmark_test.go index ab9f1c9..235379f 100644 --- a/pkg/runtime/builtin_benchmark_test.go +++ b/pkg/runtime/builtin_benchmark_test.go @@ -1,9 +1,14 @@ package runtime import ( + "context" + "io" "testing" "github.com/MontFerret/cli/v2/pkg/logger" + "github.com/MontFerret/ferret/v2/pkg/bytecode/artifact" + "github.com/MontFerret/ferret/v2/pkg/compiler" + "github.com/MontFerret/ferret/v2/pkg/source" ) func BenchmarkBuiltinLifecycle(b *testing.B) { @@ -14,7 +19,7 @@ func BenchmarkBuiltinLifecycle(b *testing.B) { b.ResetTimer() for b.Loop() { - rt, err := newBuiltin(opts) + rt, err := New(context.Background(), opts) if err != nil { b.Fatal(err) } @@ -24,3 +29,72 @@ func BenchmarkBuiltinLifecycle(b *testing.B) { } } } + +func BenchmarkBuiltinArtifactLifecycle(b *testing.B) { + c, err := compiler.New() + if err != nil { + b.Fatal(err) + } + + program, err := c.Compile(b.Context(), source.New("benchmark.fql", "RETURN @value")) + if err != nil { + b.Fatal(err) + } + + data, err := artifact.Marshal(program) + if err != nil { + b.Fatal(err) + } + + opts := NewDefaultOptions() + opts.Logger.LogOutput = logger.OutputNone + params := map[string]any{"value": 42} + b.ReportAllocs() + b.ResetTimer() + + for b.Loop() { + out, err := RunArtifact(b.Context(), opts, data, params) + if err != nil { + b.Fatal(err) + } + + if _, err := io.Copy(io.Discard, out); err != nil { + b.Fatal(err) + } + + if err := out.Close(); err != nil { + b.Fatal(err) + } + } +} + +func BenchmarkBuiltinSource(b *testing.B) { + opts := NewDefaultOptions() + opts.Logger.LogOutput = logger.OutputNone + + rt, err := New(context.Background(), opts) + if err != nil { + b.Fatal(err) + } + + b.Cleanup(func() { _ = rt.Close() }) + src := source.New("benchmark.fql", "RETURN @value") + params := map[string]any{"value": 42} + b.ReportAllocs() + b.ResetTimer() + + for b.Loop() { + out, err := RunSource(context.Background(), rt, src, params) + if err != nil { + b.Fatal(err) + } + + if _, err := io.Copy(io.Discard, out); err != nil { + b.Fatal(err) + } + + if err := out.Close(); err != nil { + b.Fatal(err) + } + } +} diff --git a/pkg/runtime/close_func_test.go b/pkg/runtime/close_func_test.go new file mode 100644 index 0000000..89ab25b --- /dev/null +++ b/pkg/runtime/close_func_test.go @@ -0,0 +1,7 @@ +package runtime + +type closeFunc func() error + +func (f closeFunc) Close() error { + return f() +} diff --git a/pkg/runtime/debug.go b/pkg/runtime/debug.go index 8fbcc0b..a598ce1 100644 --- a/pkg/runtime/debug.go +++ b/pkg/runtime/debug.go @@ -14,7 +14,7 @@ import ( type DebugSession struct { *ferret.DebugSession - runtime *Builtin + runtime *Resources plan *ferret.Plan closeErr error closeOnce sync.Once @@ -33,12 +33,12 @@ func NewDebugSession(ctx context.Context, opts Options, params map[string]any, s return nil, ErrDebugRequiresBuiltinRuntime } - rt, err := newBuiltin(opts) + rt, err := New(ctx, opts) if err != nil { return nil, err } - plan, err := rt.engine.CompileDebug(ctx, src) + plan, err := rt.builtin.engine.CompileDebug(ctx, src) if err != nil { return nil, errors.Join(err, rt.Close()) } diff --git a/pkg/runtime/factory.go b/pkg/runtime/factory.go new file mode 100644 index 0000000..748d9ad --- /dev/null +++ b/pkg/runtime/factory.go @@ -0,0 +1,116 @@ +package runtime + +import ( + "context" + "errors" + "fmt" + "io" + "net" + "net/url" + "sync/atomic" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + + "github.com/MontFerret/api" + "github.com/MontFerret/ferret/v2/uapi" + "github.com/MontFerret/wire/client" +) + +// New creates the selected runtime and owns its resources for one invocation or REPL. +// Its context bounds Wire construction; callers pass their execution contexts +// to the returned api.Runtime and settle those calls before closing the owner. +func New(ctx context.Context, opts Options) (*Resources, error) { + opts = NormalizeOptions(opts) + + if err := ValidateOptions(opts); err != nil { + return nil, err + } + + if err := ctx.Err(); err != nil { + return nil, err + } + + if IsBuiltinType(opts.Type) { + native, err := newBuiltin(opts) + if err != nil { + return nil, err + } + + rt := uapi.Wrap(native.engine, api.Version(EmbeddedVersion())) + + return &Resources{Runtime: rt, builtin: native, owners: []io.Closer{rt, native}}, nil + } + + if IsWireType(opts.Type) { + return openWire(ctx, opts) + } + + u, err := url.Parse(normalizeRuntimeType(opts.Type)) + if err != nil { + return nil, fmt.Errorf("parse url: %w", err) + } + + legacy := newRemote(*u, opts) + + return &Resources{Legacy: legacy, owners: []io.Closer{legacy}}, nil +} + +func openWire(ctx context.Context, opts Options) (*Resources, error) { + address, err := wireAddress(opts.Endpoint) + if err != nil { + return nil, err + } + + connectCtx, cancel := context.WithTimeout(ctx, opts.ConnectTimeout) + defer cancel() + var dialed atomic.Bool + + connection, err := grpc.NewClient("passthrough:///"+address, + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithNoProxy(), + grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { + // gRPC reconnects transports by default. One Wire invocation owns + // one physical connection; transport loss requires a new command. + if !dialed.CompareAndSwap(false, true) { + return nil, fmt.Errorf("wire transport connection was lost; start a new command") + } + + return new(net.Dialer).DialContext(ctx, "tcp4", address) + }), + ) + if err != nil { + return nil, fmt.Errorf("connect Wire runtime %s: %w", opts.Endpoint, err) + } + + physical := &transport{connection: connection} + // Bound stream creation too: Wire detaches the long-lived Connect stream + // from the construction context, so a stalled HTTP/2 handshake must close + // the physical transport to unblock construction. + closed := make(chan struct{}) + stop := context.AfterFunc(connectCtx, func() { + _ = physical.Close() + close(closed) + }) + rt, err := client.New(connectCtx, connection) + if !stop() { + <-closed + } + + if ctxErr := connectCtx.Err(); ctxErr != nil { + err = errors.Join(err, ctxErr) + } + + resources := &Resources{Runtime: rt, owners: []io.Closer{physical}} + if rt != nil { + resources.owners = []io.Closer{rt, physical} + } + + if err != nil { + return nil, errors.Join(fmt.Errorf("connect Wire runtime %s: %w", opts.Endpoint, err), resources.Close()) + } + + resources.Disconnected = physical.watch(connection) + + return resources, nil +} diff --git a/pkg/runtime/fs_policy_test.go b/pkg/runtime/fs_policy_test.go index 7126029..c5828f8 100644 --- a/pkg/runtime/fs_policy_test.go +++ b/pkg/runtime/fs_policy_test.go @@ -110,7 +110,7 @@ func TestBuiltinFilesystemPolicyRejectsMissingRoot(t *testing.T) { opts := NewDefaultOptions() opts.FSPolicy = &FileSystemPolicy{Root: root} - _, err := NewBuiltin(opts) + _, err := New(t.Context(), opts) if err == nil || !strings.Contains(err.Error(), root) { t.Fatalf("expected filesystem root error, got %v", err) } @@ -121,7 +121,7 @@ func TestNewRejectsFilesystemPolicyForRemoteRuntime(t *testing.T) { opts.Type = "https://worker.example" opts.FSPolicy = &FileSystemPolicy{ReadOnly: true} - _, err := New(opts) + _, err := New(t.Context(), opts) if !errors.Is(err, ErrFSPolicyRequiresBuiltinRuntime) { t.Fatalf("expected builtin runtime policy error, got %v", err) } diff --git a/pkg/runtime/http_policy_test.go b/pkg/runtime/http_policy_test.go index a5e30cb..efe79dc 100644 --- a/pkg/runtime/http_policy_test.go +++ b/pkg/runtime/http_policy_test.go @@ -65,7 +65,7 @@ func TestBuiltinHTTPPolicyRejectsInvalidConfiguration(t *testing.T) { opts := NewDefaultOptions() opts.HTTPPolicy = []ferrethttp.PolicyOption{ferrethttp.WithAllowedHosts("bad host")} - _, err := NewBuiltin(opts) + _, err := newBuiltin(opts) if !errors.Is(err, ferrethttp.ErrInvalidPolicyConfiguration) || !strings.Contains(err.Error(), "initialize HTTP policy") || !strings.Contains(err.Error(), "allowed hosts") || @@ -101,7 +101,7 @@ func TestBuiltinCloseSucceedsWithConfiguredHTTPPolicy(t *testing.T) { opts := NewDefaultOptions() opts.HTTPPolicy = []ferrethttp.PolicyOption{ferrethttp.WithAllowLocalhost(true)} - rt, err := NewBuiltin(opts) + rt, err := New(t.Context(), opts) if err != nil { t.Fatal(err) } @@ -115,7 +115,7 @@ func TestNewRejectsHTTPPolicyForRemoteRuntime(t *testing.T) { opts.Type = "https://worker.example" opts.HTTPPolicy = []ferrethttp.PolicyOption{ferrethttp.WithAllowLocalhost(true)} - _, err := New(opts) + _, err := New(t.Context(), opts) if !errors.Is(err, ErrHTTPPolicyRequiresBuiltinRuntime) { t.Fatalf("expected builtin runtime policy error, got %v", err) } diff --git a/pkg/runtime/incompatible_handshake_test.go b/pkg/runtime/incompatible_handshake_test.go new file mode 100644 index 0000000..5d2d3d6 --- /dev/null +++ b/pkg/runtime/incompatible_handshake_test.go @@ -0,0 +1,40 @@ +package runtime + +import ( + "context" + "net" + "strings" + "testing" + "time" + + "google.golang.org/grpc" + + wirev1 "github.com/MontFerret/wire/gen/ferret/wire/v1" +) + +func TestWireRejectsHandshakeWithoutRuntimeVersion(t *testing.T) { + listener, err := net.Listen("tcp4", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + + defer listener.Close() + server := grpc.NewServer() + wirev1.RegisterRuntimeServiceServer(server, &unversionedServer{}) + defer server.Stop() + served := make(chan error, 1) + go func() { served <- server.Serve(listener) }() + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + resources, err := New(ctx, Options{Type: "wire", Endpoint: "tcp://" + listener.Addr().String()}) + if resources != nil || err == nil || !strings.Contains(err.Error(), "invalid Connect handshake") { + t.Fatalf("unversioned handshake: resources=%v error=%v", resources, err) + } + + server.Stop() + select { + case <-served: + case <-ctx.Done(): + t.Fatal("incompatible server did not stop") + } +} diff --git a/pkg/runtime/metadata.go b/pkg/runtime/metadata.go new file mode 100644 index 0000000..996d514 --- /dev/null +++ b/pkg/runtime/metadata.go @@ -0,0 +1,7 @@ +package runtime + +// EmbeddedVersion returns CLI-injected metadata for the embedded native engine. +// It does not describe an external runtime. +func EmbeddedVersion() string { + return version +} diff --git a/pkg/runtime/options.go b/pkg/runtime/options.go index 54374c0..e528063 100644 --- a/pkg/runtime/options.go +++ b/pkg/runtime/options.go @@ -2,6 +2,7 @@ package runtime import ( "fmt" + "time" "github.com/MontFerret/cli/v2/pkg/logger" "github.com/MontFerret/contrib/modules/web/html/drivers" @@ -11,16 +12,25 @@ import ( ) type Options struct { - Type string - Proxy string - UserAgent string - Headers *drivers.HTTPHeaders - Cookies *drivers.HTTPCookies - KeepCookies bool - BrowserAddress string - WithBrowser bool - WithHeadlessBrowser bool - Logger logger.Options + Type string + // Endpoint and ConnectTimeout configure Wire connections for source and metadata commands. + Endpoint string + EndpointSet bool + ConnectTimeout time.Duration + ConnectTimeoutSet bool + // BrowserOptionsSet distinguishes explicitly configured browser settings from defaults. + BrowserOptionsSet bool + // BuiltinPolicyOptionsSet records explicit policy configuration even in metadata commands. + BuiltinPolicyOptionsSet bool + Proxy string + UserAgent string + Headers *drivers.HTTPHeaders + Cookies *drivers.HTTPCookies + KeepCookies bool + BrowserAddress string + WithBrowser bool + WithHeadlessBrowser bool + Logger logger.Options // FSPolicy configures filesystem access for the builtin runtime only. FSPolicy *FileSystemPolicy // HTTPPolicy configures outbound HTTP for the builtin runtime only. @@ -36,6 +46,7 @@ type FileSystemPolicy struct { func NewDefaultOptions() Options { return Options{ Type: DefaultRuntime, + ConnectTimeout: DefaultConnectTimeout, BrowserAddress: cdp.DefaultAddress, Proxy: "", UserAgent: "", @@ -69,12 +80,36 @@ func ValidateOptions(opts Options) error { return ErrFSPolicyRequiresBuiltinRuntime } + if IsWireType(opts.Type) { + if opts.BuiltinPolicyOptionsSet { + return fmt.Errorf("wire mode does not accept filesystem or HTTP policies; configure the runtime host") + } + + if _, err := wireAddress(opts.Endpoint); err != nil { + return err + } + + if opts.ConnectTimeout <= 0 { + return fmt.Errorf("--runtime-connect-timeout must be positive") + } + + if opts.BrowserOptionsSet || opts.WithBrowser || opts.WithHeadlessBrowser || opts.KeepCookies || opts.Proxy != "" || opts.UserAgent != "" || opts.Headers != nil || opts.Cookies != nil || (opts.BrowserAddress != "" && opts.BrowserAddress != DefaultBrowser && opts.BrowserAddress != cdp.DefaultAddress) { + return fmt.Errorf("wire mode does not accept browser configuration; configure the runtime host") + } + } else if opts.EndpointSet || opts.Endpoint != "" || opts.ConnectTimeoutSet { + return fmt.Errorf("--runtime-endpoint and --runtime-connect-timeout require --runtime wire") + } + return nil } func NormalizeOptions(opts Options) Options { opts.Logger = logger.NormalizeOptions(opts.Logger) + if opts.ConnectTimeout == 0 && !opts.ConnectTimeoutSet { + opts.ConnectTimeout = DefaultConnectTimeout + } + return opts } diff --git a/pkg/runtime/remote.go b/pkg/runtime/remote.go index daf5a43..dee13c5 100644 --- a/pkg/runtime/remote.go +++ b/pkg/runtime/remote.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -35,7 +36,7 @@ type ( } ) -func NewRemote(url url.URL, opts Options) Runtime { +func newRemote(url url.URL, opts Options) *Remote { rt := new(Remote) rt.url = url rt.opts = opts @@ -44,13 +45,15 @@ func NewRemote(url url.URL, opts Options) Runtime { return rt } -func (rt *Remote) Version(ctx context.Context) (string, error) { +func (rt *Remote) Version(ctx context.Context) (version string, err error) { data, err := rt.makeRequest(ctx, "GET", "/info", nil) if err != nil { return "", fmt.Errorf("make request: %w", err) } + defer func() { err = errors.Join(err, data.Close()) }() + info := remoteInfo{} b, err := io.ReadAll(data) @@ -79,10 +82,6 @@ func (rt *Remote) Run(ctx context.Context, query source.Source, params map[strin return rt.makeRequest(ctx, "POST", "/", body) } -func (rt *Remote) RunArtifact(_ context.Context, _ []byte, _ map[string]any) (io.ReadCloser, error) { - return nil, ErrArtifactRequiresBuiltinRuntime -} - func (rt *Remote) Close() error { return nil } diff --git a/pkg/runtime/remote_version_test.go b/pkg/runtime/remote_version_test.go new file mode 100644 index 0000000..e6e73ee --- /dev/null +++ b/pkg/runtime/remote_version_test.go @@ -0,0 +1,39 @@ +package runtime + +import ( + "errors" + "io" + "net/http" + "net/url" + "strings" + "testing" +) + +func TestLegacyVersionClosesResponseAndRetainsFailure(t *testing.T) { + closeErr := errors.New("close version response failed") + for _, content := range []string{`{"version":{"ferret":"legacy"}}`, "invalid JSON"} { + closed := 0 + body := &struct { + io.Reader + io.Closer + }{strings.NewReader(content), closeFunc(func() error { + closed++ + + return closeErr + })} + rt := &Remote{url: url.URL{Scheme: "http", Host: "worker.invalid"}, client: &http.Client{ + Transport: roundTripFunc(func(*http.Request) (*http.Response, error) { + return &http.Response{StatusCode: http.StatusOK, Body: body}, nil + }), + }} + + version, err := rt.Version(t.Context()) + if closed != 1 || !errors.Is(err, closeErr) { + t.Fatalf("version=%q closed=%d error=%v", version, closed, err) + } + + if content == "invalid JSON" && !strings.Contains(err.Error(), "deserialize response data") { + t.Fatalf("construction failure lost: %v", err) + } + } +} diff --git a/pkg/runtime/resources.go b/pkg/runtime/resources.go new file mode 100644 index 0000000..47446b0 --- /dev/null +++ b/pkg/runtime/resources.go @@ -0,0 +1,46 @@ +package runtime + +import ( + "errors" + "fmt" + "io" + "sync" + + "github.com/MontFerret/api" +) + +// Resources owns the selected runtime and the resources created for it. +// Runtime is canonical UAPI for builtin/Wire; Legacy is set only for HTTP Worker +// compatibility. Callers must settle active calls and close any directly created +// children before Close. The exported runtime fields are borrowed by callers. +type Resources struct { + Runtime api.Runtime + Legacy *Remote + // Disconnected closes on loss of an established Wire transport. It is nil + // for builtin and HTTP. Callers may use it to interrupt idle input, but must + // still settle runtime calls before closing these resources. + Disconnected <-chan struct{} + builtin *Builtin + owners []io.Closer + once sync.Once + err error +} + +// Close releases the runtime before its native engine or physical transport. +// It is safe to repeat and retains all cleanup failures. Wire performs bounded +// detached logical cleanup even after the execution context has been cancelled. +func (r *Resources) Close() error { + if r == nil { + return nil + } + + r.once.Do(func() { + for _, owner := range r.owners { + if err := owner.Close(); err != nil { + r.err = errors.Join(r.err, fmt.Errorf("close runtime resource: %w", err)) + } + } + }) + + return r.err +} diff --git a/pkg/runtime/round_trip_func_test.go b/pkg/runtime/round_trip_func_test.go new file mode 100644 index 0000000..9bfa8ab --- /dev/null +++ b/pkg/runtime/round_trip_func_test.go @@ -0,0 +1,9 @@ +package runtime + +import "net/http" + +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (f roundTripFunc) RoundTrip(request *http.Request) (*http.Response, error) { + return f(request) +} diff --git a/pkg/runtime/runtime.go b/pkg/runtime/runtime.go index 6bab662..3ca2c37 100644 --- a/pkg/runtime/runtime.go +++ b/pkg/runtime/runtime.go @@ -5,44 +5,13 @@ import ( "errors" "fmt" "io" - "net/url" "strings" "github.com/MontFerret/ferret/v2/pkg/source" ) -type Runtime interface { - Version(ctx context.Context) (string, error) - - Run(ctx context.Context, query source.Source, params map[string]any) (io.ReadCloser, error) - RunArtifact(ctx context.Context, data []byte, params map[string]any) (io.ReadCloser, error) - Close() error -} - -func New(opts Options) (Runtime, error) { - opts = NormalizeOptions(opts) - - if err := ValidateOptions(opts); err != nil { - return nil, err - } - - name := normalizeRuntimeType(opts.Type) - - if IsBuiltinType(name) { - return NewBuiltin(opts) - } - - u, err := url.Parse(name) - - if err != nil { - return nil, fmt.Errorf("parse url: %w", err) - } - - return NewRemote(*u, opts), nil -} - func Run(ctx context.Context, opts Options, query source.Source, params map[string]any) (out io.ReadCloser, err error) { - rt, err := New(opts) + rt, err := New(ctx, opts) if err != nil { return nil, err @@ -54,11 +23,15 @@ func Run(ctx context.Context, opts Options, query source.Source, params map[stri } }() - return rt.Run(ctx, query, params) + return RunSource(ctx, rt, query, params) } func RunArtifact(ctx context.Context, opts Options, data []byte, params map[string]any) (out io.ReadCloser, err error) { - rt, err := New(opts) + if !IsBuiltinType(opts.Type) { + return nil, ErrArtifactRequiresBuiltinRuntime + } + + rt, err := New(ctx, opts) if err != nil { return nil, err @@ -70,7 +43,7 @@ func RunArtifact(ctx context.Context, opts Options, data []byte, params map[stri } }() - return rt.RunArtifact(ctx, data, params) + return rt.builtin.RunArtifact(ctx, data, params) } func IsBuiltinType(name string) bool { diff --git a/pkg/runtime/runtime_test.go b/pkg/runtime/runtime_test.go index ea44145..e549a3b 100644 --- a/pkg/runtime/runtime_test.go +++ b/pkg/runtime/runtime_test.go @@ -3,7 +3,11 @@ package runtime import ( "context" "errors" + "strings" "testing" + + "github.com/MontFerret/api" + "github.com/MontFerret/ferret/v2/pkg/source" ) func TestRunArtifact_RemoteRuntimeRejected(t *testing.T) { @@ -17,3 +21,61 @@ func TestRunArtifact_RemoteRuntimeRejected(t *testing.T) { t.Fatalf("unexpected error: %v", err) } } + +func TestNewRejectsCanceledConstruction(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + cancel() + + for _, kind := range []string{DefaultRuntime, "wire", "https://worker.example"} { + t.Run(kind, func(t *testing.T) { + opts := Options{Type: kind} + if kind == "wire" { + opts.Endpoint = "tcp://127.0.0.1:123" + } + + resources, err := New(ctx, opts) + if resources != nil || !errors.Is(err, context.Canceled) { + t.Fatalf("constructed resources after cancellation: %v, %v", resources, err) + } + }) + } +} + +func TestResourcesOwnsBuiltinEngine(t *testing.T) { + resources, err := New(t.Context(), NewDefaultOptions()) + if err != nil { + t.Fatal(err) + } + + t.Cleanup(func() { _ = resources.Close() }) + + // The borrowed UAPI adapter's Close leaves the native engine usable. + // Resources.Close must still release that engine afterward. + if err := resources.Runtime.Close(); err != nil { + t.Fatal(err) + } + + output, err := resources.Runtime.Run(t.Context(), api.NewSource("owned.fql", "RETURN 42")) + if err != nil || output == nil || string(output.Content) != "42" { + t.Fatalf("borrowed adapter became unusable: output=%v error=%v", output, err) + } + + nativePlan, err := resources.builtin.engine.Compile(t.Context(), source.New("survivor.fql", "RETURN 42")) + if err != nil { + t.Fatalf("adapter closed the borrowed engine: %v", err) + } + + if err := nativePlan.Close(); err != nil { + t.Fatal(err) + } + + for range 2 { + if err := resources.Close(); err != nil { + t.Fatal(err) + } + } + + if _, err := resources.builtin.engine.Compile(t.Context(), source.New("closed.fql", "RETURN 42")); err == nil || !strings.Contains(err.Error(), "engine is closed") { + t.Fatalf("resource owner left its native engine usable: %v", err) + } +} diff --git a/pkg/runtime/scripted_runtime_test.go b/pkg/runtime/scripted_runtime_test.go new file mode 100644 index 0000000..8aa9378 --- /dev/null +++ b/pkg/runtime/scripted_runtime_test.go @@ -0,0 +1,33 @@ +package runtime + +import ( + "context" + + "github.com/MontFerret/api" +) + +type scriptedRuntime struct { + api.Runtime + output *api.Output + err error + closeErr error + closed int +} + +func (r *scriptedRuntime) Version(ctx context.Context) (api.Version, error) { + if err := ctx.Err(); err != nil { + return "", err + } + + return "scripted-runtime", nil +} + +func (r *scriptedRuntime) Run(context.Context, api.Source, ...api.SessionOption) (*api.Output, error) { + return r.output, r.err +} + +func (r *scriptedRuntime) Close() error { + r.closed++ + + return r.closeErr +} diff --git a/pkg/runtime/source_execution.go b/pkg/runtime/source_execution.go new file mode 100644 index 0000000..07eb4b8 --- /dev/null +++ b/pkg/runtime/source_execution.go @@ -0,0 +1,30 @@ +package runtime + +import ( + "bytes" + "context" + "io" + + "github.com/MontFerret/api" + "github.com/MontFerret/ferret/v2/pkg/source" +) + +// RunSource executes through the selected canonical runtime or isolated Worker +// adapter. Output remains available alongside an execution or cleanup error. +func RunSource(ctx context.Context, resources *Resources, src source.Source, params map[string]any) (io.ReadCloser, error) { + if resources.Legacy != nil { + return resources.Legacy.Run(ctx, src, params) + } + + output, err := resources.Runtime.Run(ctx, api.NewSource(src.Name(), src.Content()), api.WithParams(params)) + + return outputReader(output), err +} + +func outputReader(output *api.Output) io.ReadCloser { + if output == nil { + return nil + } + + return io.NopCloser(bytes.NewReader(output.Content)) +} diff --git a/pkg/runtime/source_test.go b/pkg/runtime/source_test.go new file mode 100644 index 0000000..857084b --- /dev/null +++ b/pkg/runtime/source_test.go @@ -0,0 +1,273 @@ +package runtime + +import ( + "context" + "errors" + "io" + "net" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "google.golang.org/grpc" + + "github.com/MontFerret/api" + "github.com/MontFerret/cli/v2/internal/testutil/wirehost" + "github.com/MontFerret/ferret/v2/pkg/source" +) + +func TestWireEndpointValidation(t *testing.T) { + for _, endpoint := range []string{"", "tcp://localhost:123", "tcp://127.0.0.2:123", "tcp://[::1]:123", "http://127.0.0.1:123", "tcp://user@127.0.0.1:123", "tcp://127.0.0.1:0", "tcp://127.0.0.1:65536", "tcp://127.0.0.1:-1", "tcp://127.0.0.1:123/", "tcp://127.0.0.1:123?", "tcp://127.0.0.1:123#", "tcp://127.0.0.1:123/path", " tcp://127.0.0.1:123"} { + t.Run(endpoint, func(t *testing.T) { + if _, err := wireAddress(endpoint); err == nil { + t.Fatal("accepted invalid endpoint") + } + }) + } + + for _, endpoint := range []string{"tcp://127.0.0.1:1", "tcp://127.0.0.1:65535"} { + if _, err := wireAddress(endpoint); err != nil { + t.Fatal(err) + } + } +} + +func TestSourceOutputAndCleanupFailures(t *testing.T) { + runErr := errors.New("execution failed") + closeErr := errors.New("logical cleanup failed") + physicalErr := errors.New("physical cleanup failed") + + for _, output := range []*api.Output{nil, {}, {Content: []byte("available")}} { + rt := &scriptedRuntime{output: output, err: runErr, closeErr: closeErr} + physicalClosed := 0 + resources := &Resources{Runtime: rt, owners: []io.Closer{rt, closeFunc(func() error { + if rt.closed != 1 { + t.Error("transport closed before runtime") + } + physicalClosed++ + + return physicalErr + })}} + reader, err := RunSource(t.Context(), resources, source.New("source.fql", "RETURN 42"), nil) + if !errors.Is(err, runErr) || (reader == nil) != (output == nil) { + t.Fatalf("output presence lost: reader=%v error=%v", reader, err) + } + + if reader != nil { + data, _ := io.ReadAll(reader) + _ = reader.Close() + if string(data) != string(output.Content) { + t.Fatalf("output changed: %q", data) + } + } + + for range 2 { + err = resources.Close() + if !errors.Is(err, closeErr) || !errors.Is(err, physicalErr) { + t.Fatalf("cleanup errors lost: %v", err) + } + } + + if rt.closed != 1 || physicalClosed != 1 { + t.Fatal("cleanup repeated") + } + } + + rt := &scriptedRuntime{output: &api.Output{}} + reader, err := RunSource(t.Context(), &Resources{Runtime: rt}, source.New("empty.fql", "RETURN 42"), nil) + if err != nil || reader == nil { + t.Fatalf("successful empty output lost: %v", err) + } + + data, _ := io.ReadAll(reader) + _ = reader.Close() + if len(data) != 0 { + t.Fatalf("manufactured output: %q", data) + } +} + +func TestWireFailedHandshake(t *testing.T) { + listener, err := net.Listen("tcp4", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + + server := grpc.NewServer() + defer server.Stop() + defer listener.Close() + served := make(chan error, 1) + go func() { served <- server.Serve(listener) }() + resources, err := New(t.Context(), Options{Type: "wire", Endpoint: "tcp://" + listener.Addr().String()}) + if resources != nil || err == nil || !strings.Contains(err.Error(), "unknown service") { + t.Fatalf("failed handshake: resources=%v error=%v", resources, err) + } + + server.Stop() + select { + case <-served: + case <-time.After(10 * time.Second): + t.Fatal("server did not stop") + } +} + +func TestWireConfiguredRuntimeAndOwnership(t *testing.T) { + host := wirehost.New(t) + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + resources, err := New(ctx, Options{Type: "wire", Endpoint: host.Endpoint}) + if err != nil { + t.Fatal(err) + } + + if resources.builtin != nil || resources.Legacy != nil { + t.Fatal("Wire construction created a local or legacy runtime") + } + + for _, value := range []any{nil, true, 42.0, "123", []any{1.0, false}, map[string]any{"value": nil}} { + output, err := resources.Runtime.Run(ctx, api.NewSource("host.fql", "RETURN DEMO::IDENTITY(@value)"), api.WithParams(map[string]any{"value": value})) + if err != nil || output == nil { + t.Fatalf("run %v: %v", value, err) + } + } + + if host.Connections() != 1 { + t.Fatalf("connections = %d, want 1", host.Connections()) + } + + for range 2 { + if err := resources.Close(); err != nil { + t.Fatal(err) + } + } + + if _, err := host.Runtime.Run(ctx, api.NewSource("still-owned.fql", "RETURN DEMO::IDENTITY(42)")); err != nil { + t.Fatalf("client closed host runtime: %v", err) + } +} + +func TestWireExecutionCancellation(t *testing.T) { + host := wirehost.New(t) + ctx, cancel := context.WithCancel(t.Context()) + resources, err := New(ctx, Options{Type: "wire", Endpoint: host.Endpoint}) + if err != nil { + t.Fatal(err) + } + + done := make(chan error, 1) + go func() { + _, err := resources.Runtime.Run(ctx, api.NewSource("wait.fql", "RETURN DEMO::WAIT()")) + done <- err + }() + waitSignal(t, host.WaitStarted) + cancel() + + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("cancellation error = %v", err) + } + case <-time.After(10 * time.Second): + t.Fatal("execution did not settle") + } + + waitSignal(t, host.WaitFinished) + if err := resources.Close(); err != nil { + t.Fatal(err) + } +} + +func TestWireHandshakeTimeoutAndCancellation(t *testing.T) { + for _, cancelParent := range []bool{false, true} { + t.Run(map[bool]string{false: "timeout", true: "parent cancellation"}[cancelParent], func(t *testing.T) { + listener, err := net.Listen("tcp4", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + + defer listener.Close() + accepted := make(chan net.Conn, 1) + go func() { connection, _ := listener.Accept(); accepted <- connection }() + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + done := make(chan error, 1) + go func() { + _, err := New(ctx, Options{Type: "wire", Endpoint: "tcp://" + listener.Addr().String(), ConnectTimeout: 200 * time.Millisecond}) + done <- err + }() + + var connection net.Conn + select { + case connection = <-accepted: + if connection == nil { + t.Fatal("accept failed") + } + case <-time.After(10 * time.Second): + t.Fatal("connection not established") + } + + defer connection.Close() + if cancelParent { + cancel() + } + + select { + case err := <-done: + want := context.DeadlineExceeded + if cancelParent { + want = context.Canceled + } + if !errors.Is(err, want) || !strings.Contains(err.Error(), listener.Addr().String()) { + t.Fatalf("handshake error = %v", err) + } + case <-time.After(10 * time.Second): + t.Fatal("handshake did not stop") + } + + _ = connection.SetReadDeadline(time.Now().Add(time.Second)) + if _, err := io.Copy(io.Discard, connection); err != nil { + t.Fatalf("client transport leaked: %v", err) + } + }) + } +} + +func TestLegacyHTTPExecution(t *testing.T) { + var requests atomic.Int32 + worker := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "POST" || r.URL.Path != "/" { + t.Errorf("unexpected request: %s %s", r.Method, r.URL) + } + + body, _ := io.ReadAll(r.Body) + if string(body) != `{"text":"RETURN @value","params":{"value":42}}` { + t.Errorf("Worker request = %s", body) + } + + requests.Add(1) + _, _ = io.WriteString(w, "42") + })) + defer worker.Close() + output, err := Run(t.Context(), Options{Type: worker.URL}, source.New("legacy.fql", "RETURN @value"), map[string]any{"value": 42}) + if err != nil { + t.Fatal(err) + } + + assertRuntimeOutput(t, output, "42") + + if got := requests.Load(); got != 1 { + t.Fatalf("legacy execution made %d HTTP Worker requests, want 1", got) + } +} + +func waitSignal(t *testing.T, signal <-chan struct{}) { + t.Helper() + + select { + case <-signal: + case <-time.After(10 * time.Second): + t.Fatal("fixture did not reach expected state") + } +} diff --git a/pkg/runtime/transport.go b/pkg/runtime/transport.go new file mode 100644 index 0000000..4cbeb40 --- /dev/null +++ b/pkg/runtime/transport.go @@ -0,0 +1,56 @@ +package runtime + +import ( + "context" + "io" + "sync" + + "google.golang.org/grpc" + "google.golang.org/grpc/connectivity" +) + +// transport retains one physical close shared by construction cancellation and cleanup. +type transport struct { + connection io.Closer + once sync.Once + err error + stopWatch context.CancelFunc + watchDone chan struct{} +} + +// watch observes the established transport without probing or making RPCs. +// A loss is terminal for the logical Wire runtime; this never reconnects it. +func (t *transport) watch(connection *grpc.ClientConn) <-chan struct{} { + ctx, cancel := context.WithCancel(context.Background()) + t.stopWatch = cancel + t.watchDone = make(chan struct{}) + lost := make(chan struct{}) + go func() { + defer close(t.watchDone) + + for connection.GetState() == connectivity.Ready { + if !connection.WaitForStateChange(ctx, connectivity.Ready) { + return + } + } + + if ctx.Err() == nil { + close(lost) + } + }() + + return lost +} + +func (t *transport) Close() error { + t.once.Do(func() { + if t.stopWatch != nil { + t.stopWatch() + <-t.watchDone + } + + t.err = t.connection.Close() + }) + + return t.err +} diff --git a/pkg/runtime/uapi_metadata_test.go b/pkg/runtime/uapi_metadata_test.go new file mode 100644 index 0000000..2bb61c0 --- /dev/null +++ b/pkg/runtime/uapi_metadata_test.go @@ -0,0 +1,268 @@ +package runtime + +import ( + "context" + "errors" + "slices" + "sync/atomic" + "testing" + "time" + + "github.com/MontFerret/api" + "github.com/MontFerret/cli/v2/internal/testutil/wirehost" +) + +func TestSourceRuntimeVersion(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + + builtin, err := New(ctx, NewDefaultOptions()) + if err != nil { + t.Fatal(err) + } + + t.Cleanup(func() { + if err := builtin.Close(); err != nil { + t.Error(err) + } + }) + got, err := builtin.Runtime.Version(ctx) + if err != nil || got.String() != EmbeddedVersion() { + t.Fatalf("builtin version=%q, %v; want %q", got, err, EmbeddedVersion()) + } + + for _, value := range []api.Version{"v2.0.0-alpha.57", " runtime+opaque \n", "版本-α", "runtime\x00\xff", ""} { + t.Run(string(value), func(t *testing.T) { + host := wirehost.NewWithVersion(t, value) + resources, err := New(ctx, Options{Type: "wire", Endpoint: host.Endpoint}) + if err != nil { + t.Fatal(err) + } + + t.Cleanup(func() { + if err := resources.Close(); err != nil { + t.Error(err) + } + }) + for range 2 { + got, err := resources.Runtime.Version(ctx) + if err != nil || got != value { + t.Fatalf("hosted version=%q, %v; want %q", got, err, value) + } + } + + if err := resources.Close(); err != nil { + t.Fatal(err) + } + + got, err := resources.Runtime.Version(ctx) + if err != nil || got != value || host.Connections() != 1 { + t.Fatalf("metadata after cleanup=%q, %v; connections=%d", got, err, host.Connections()) + } + }) + } +} + +func TestSourcePlanParameterMetadata(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + + for _, mode := range []string{"builtin", "wire"} { + t.Run(mode, func(t *testing.T) { + opts := NewDefaultOptions() + if mode == "wire" { + host := wirehost.New(t) + opts.Type, opts.Endpoint = mode, host.Endpoint + } + + resources, err := New(ctx, opts) + if err != nil { + t.Fatal(err) + } + + t.Cleanup(func() { + if err := resources.Close(); err != nil { + t.Error(err) + } + }) + canceled, cancel := context.WithCancel(ctx) + cancel() + expired, expire := context.WithDeadline(ctx, time.Now().Add(-time.Second)) + defer expire() + + type parameterSnapshot struct { + plan api.Plan + params []string + } + + var plans []parameterSnapshot + + for _, debug := range []bool{false, true} { + for _, query := range []string{"RETURN 1", "RETURN [@second, @first, @second]"} { + compile := resources.Runtime.Compile + if debug { + compile = resources.Runtime.CompileDebug + } + + plan, err := compile(ctx, api.NewSource("params.fql", query)) + if err != nil { + t.Fatal(err) + } + + t.Cleanup(func() { + if err := plan.Close(); err != nil { + t.Error(err) + } + }) + + params, err := plan.Params(ctx) + if err != nil { + t.Fatal(err) + } + + want := []string(nil) + if query != "RETURN 1" { + want = []string{"first", "second"} + } + + snapshot := slices.Clone(params) + plans = append(plans, parameterSnapshot{plan, snapshot}) + slices.Sort(params) + if !slices.Equal(params, want) { + t.Fatalf("Params=%v, want %v", params, want) + } + + if len(params) != 0 { + params[0] = "caller mutation" + } + + again, err := plan.Params(ctx) + if err != nil || !slices.Equal(again, snapshot) { + t.Fatalf("parameter snapshot changed=%v, %v; want %v", again, err, snapshot) + } + + for _, check := range []struct { + ctx context.Context + err error + }{{nil, nil}, {canceled, context.Canceled}, {expired, context.DeadlineExceeded}} { + if _, err := plan.Params(check.ctx); err == nil || (check.err != nil && !errors.Is(err, check.err)) { + t.Fatalf("Params context error=%v; want %v", err, check.err) + } + + if _, err := resources.Runtime.Version(check.ctx); err == nil || (check.err != nil && !errors.Is(err, check.err)) { + t.Fatalf("Version context error=%v; want %v", err, check.err) + } + } + + if err := plan.Close(); err != nil { + t.Fatal(err) + } + } + } + + if err := resources.Close(); err != nil { + t.Fatal(err) + } + + for _, snapshot := range plans { + params, err := snapshot.plan.Params(ctx) + if err != nil || !slices.Equal(params, snapshot.params) { + t.Fatalf("metadata after cleanup=%v, %v; want %v", params, err, snapshot.params) + } + } + }) + } +} + +func TestWireVersionHandshakeCancellation(t *testing.T) { + for _, timedOut := range []bool{false, true} { + t.Run(map[bool]string{false: "parent cancellation", true: "connect timeout"}[timedOut], func(t *testing.T) { + started, finished := make(chan struct{}), make(chan struct{}) + host := wirehost.NewWithVersionFunc(t, func(ctx context.Context) (api.Version, error) { + close(started) + defer close(finished) + <-ctx.Done() + + return "", ctx.Err() + }) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + opts := Options{Type: "wire", Endpoint: host.Endpoint, ConnectTimeout: 10 * time.Second} + if timedOut { + opts.ConnectTimeout = 250 * time.Millisecond + } + + done := make(chan error, 1) + go func() { + resources, err := New(ctx, opts) + if resources != nil { + err = errors.Join(err, errors.New("canceled handshake returned resources"), resources.Close()) + } + + done <- err + }() + waitSignal(t, started) + + if !timedOut { + cancel() + } + + select { + case err := <-done: + want := context.Canceled + if timedOut { + want = context.DeadlineExceeded + } + + if !errors.Is(err, want) { + t.Fatalf("metadata construction error=%v; want %v", err, want) + } + case <-time.After(10 * time.Second): + t.Fatal("hosted Version did not unblock construction") + } + + waitSignal(t, finished) + cleanupCtx, cleanupCancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cleanupCancel() + + if err := host.WaitForClientsClosed(cleanupCtx); err != nil { + t.Fatalf("partial construction leaked the transport: %v", err) + } + }) + } +} + +func TestWireVersionSnapshotAfterTransportLoss(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + + var calls atomic.Int32 + host := wirehost.NewWithVersionFunc(t, func(context.Context) (api.Version, error) { + calls.Add(1) + + return "hosted-opaque-version", nil + }) + resources, err := New(ctx, Options{Type: "wire", Endpoint: host.Endpoint}) + if err != nil { + t.Fatal(err) + } + + host.StopTransport() + waitSignal(t, resources.Disconnected) + + for range 2 { + version, err := resources.Runtime.Version(ctx) + if err != nil || version != "hosted-opaque-version" { + t.Fatalf("cached metadata after transport loss=%q, %v", version, err) + } + } + + first, second := resources.Close(), resources.Close() + if (first == nil) != (second == nil) || (first != nil && first.Error() != second.Error()) { + t.Fatalf("cleanup result was not retained: %v / %v", first, second) + } + + if calls.Load() != 1 || host.Connections() != 1 { + t.Fatalf("metadata reconnected or queried the host: calls=%d connections=%d", calls.Load(), host.Connections()) + } +} diff --git a/pkg/runtime/unversioned_server_test.go b/pkg/runtime/unversioned_server_test.go new file mode 100644 index 0000000..d71e9b7 --- /dev/null +++ b/pkg/runtime/unversioned_server_test.go @@ -0,0 +1,25 @@ +package runtime + +import ( + "google.golang.org/grpc" + + wirev1 "github.com/MontFerret/wire/gen/ferret/wire/v1" +) + +// unversionedServer simulates a pre-alpha.2 handshake with valid protocol metadata. +type unversionedServer struct { + wirev1.UnimplementedRuntimeServiceServer +} + +func (s *unversionedServer) Connect(_ *wirev1.ConnectRequest, stream grpc.ServerStreamingServer[wirev1.ConnectResponse]) error { + if err := stream.Send(&wirev1.ConnectResponse{ + ConnectionId: &wirev1.ConnectionId{Value: "legacy-connection"}, + Protocol: &wirev1.ProtocolInfo{Name: "ferret-wire", Version: "1"}, + }); err != nil { + return err + } + + <-stream.Context().Done() + + return stream.Context().Err() +} diff --git a/pkg/runtime/wire_benchmark_test.go b/pkg/runtime/wire_benchmark_test.go new file mode 100644 index 0000000..110ec70 --- /dev/null +++ b/pkg/runtime/wire_benchmark_test.go @@ -0,0 +1,65 @@ +package runtime + +import ( + "context" + "testing" + + "github.com/MontFerret/api" + "github.com/MontFerret/cli/v2/internal/testutil/wirehost" +) + +func BenchmarkWireConnection(b *testing.B) { + host := wirehost.New(b) + opts := Options{Type: "wire", Endpoint: host.Endpoint} + b.ReportAllocs() + b.ResetTimer() + + for b.Loop() { + resources, err := New(context.Background(), opts) + if err != nil { + b.Fatal(err) + } + + if err := resources.Close(); err != nil { + b.Fatal(err) + } + } +} + +func BenchmarkWireSource(b *testing.B) { + host := wirehost.New(b) + resources, err := New(context.Background(), Options{Type: "wire", Endpoint: host.Endpoint}) + if err != nil { + b.Fatal(err) + } + + b.Cleanup(func() { _ = resources.Close() }) + src := api.NewSource("benchmark.fql", "RETURN DEMO::IDENTITY(@value)") + params := api.WithParams(map[string]any{"value": 42}) + b.ReportAllocs() + b.ResetTimer() + + for b.Loop() { + if _, err := resources.Runtime.Run(context.Background(), src, params); err != nil { + b.Fatal(err) + } + } +} + +func BenchmarkWireVersion(b *testing.B) { + host := wirehost.New(b) + resources, err := New(context.Background(), Options{Type: "wire", Endpoint: host.Endpoint}) + if err != nil { + b.Fatal(err) + } + + b.Cleanup(func() { _ = resources.Close() }) + b.ReportAllocs() + b.ResetTimer() + + for b.Loop() { + if _, err := resources.Runtime.Version(context.Background()); err != nil { + b.Fatal(err) + } + } +} diff --git a/pkg/runtime/wire_endpoint.go b/pkg/runtime/wire_endpoint.go new file mode 100644 index 0000000..d6f4341 --- /dev/null +++ b/pkg/runtime/wire_endpoint.go @@ -0,0 +1,50 @@ +package runtime + +import ( + "fmt" + "net" + "strconv" + "strings" + "time" +) + +// DefaultConnectTimeout bounds Wire transport establishment and its logical handshake. +const DefaultConnectTimeout = 5 * time.Second + +// IsWireType identifies the explicit Wire source-execution mode. +func IsWireType(name string) bool { + return normalizeRuntimeType(name) == "wire" +} + +func wireAddress(endpoint string) (string, error) { + const prefix = "tcp://127.0.0.1:" + if endpoint == "" { + return "", fmt.Errorf("wire runtime requires --runtime-endpoint tcp://127.0.0.1:") + } + + invalid := func() (string, error) { + return "", fmt.Errorf("invalid Wire endpoint %q: use tcp://127.0.0.1: with a port from 1 to 65535", endpoint) + } + + if !strings.HasPrefix(endpoint, prefix) { + return invalid() + } + + port := strings.TrimPrefix(endpoint, prefix) + if port == "" { + return invalid() + } + + for _, char := range port { + if char < '0' || char > '9' { + return invalid() + } + } + + number, err := strconv.ParseUint(port, 10, 16) + if err != nil || number == 0 { + return invalid() + } + + return net.JoinHostPort("127.0.0.1", strconv.FormatUint(number, 10)), nil +}