diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5e246e8..277fe8c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -30,7 +30,7 @@ jobs: ${{ matrix.module }}/go.sum - run: go test -race -covermode=atomic -coverprofile=cover.out ./... - run: go vet ./... - - if: matrix.module == 'dbkit/sqlite' || matrix.module == 'gonsole/auth/sqlite' + - if: matrix.module == 'dbkit/sqlite' || matrix.module == 'gonsole/auth/sqlite' || matrix.module == 'gonsole' run: GOOS=darwin GOARCH=arm64 go vet ./... && GOOS=windows GOARCH=amd64 go vet ./... test-postgres: diff --git a/gonsole/CHANGELOG.md b/gonsole/CHANGELOG.md index 20b64d1..5a01397 100644 --- a/gonsole/CHANGELOG.md +++ b/gonsole/CHANGELOG.md @@ -8,6 +8,21 @@ v0.x, minor releases may contain breaking changes. Releases of this module are tagged `gonsole/vX.Y.Z`. +## [Unreleased] + +### Added + +- `Env.Listening` and `Listening`, where a server listens from the `SOCKET`, `ADDR` and `PORT` settings, the first set winning, with `CLIENT_ADDR_HEADER` and `IDLE_EXIT`. +- `ServeListening`, which serves like `Serve` on a TCP address or on a new Unix socket of mode 0600, sockets on Linux only. +- On a socket, `ServeListening` puts the visitor's address from the `CLIENT_ADDR_HEADER` header in `RemoteAddr`, and answers 400 when that header holds no single address. +- `ServeListening` drops the `CLIENT_ADDR_HEADER` header from every request, TCP included. +- `ServeListening` ends the run with no error once nothing has been open for `IDLE_EXIT`, so `serve` exits 0. +- `ServeListening` takes an idle exit only when the server's `ReadTimeout` stands above zero and its `IdleTimeout` and `ReadHeaderTimeout` are not negative. +- `ServeListening` counts a connection a handler takes over, such as a WebSocket, until it closes, so the idle exit waits for it. +- On a socket, `ServeListening` holds a lock on `.lock` while the socket exists, so two starts at once never remove each other's socket. +- A start that finds that lock taken stops with `a server already listens on ` when a server answers there, else with `another server is starting on `. +- `OverSocket`, whether a request came over a Unix socket. + ## [0.7.0] - 2026-10-09 ### Added diff --git a/gonsole/exec_linux_test.go b/gonsole/exec_linux_test.go new file mode 100644 index 0000000..3099176 --- /dev/null +++ b/gonsole/exec_linux_test.go @@ -0,0 +1,51 @@ +// SPDX-License-Identifier: Apache-2.0 + +//go:build linux + +package gonsole_test + +import ( + "syscall" + "testing" + + "github.com/gopherium/framework/gonsole" +) + +func TestMainServesOnASocket(t *testing.T) { + t.Parallel() + + path := socketAt(t) + variables := []string{"MYAPP_SOCKET=" + string(path), "MYAPP_CLIENT_ADDR_HEADER=X-Client-Addr"} + cmd := exampleCommand(t, variables, "serve") + lines := launch(t, cmd) + if address := listeningAddress(t, lines); address != string(path) { + t.Fatalf("listening on %q, want the socket's path %q", address, path) + } + + body := path.get(t) + signalled(t, cmd, lines, syscall.SIGTERM) + err := cmd.Wait() + + if err != nil { + t.Errorf("exit = %v, want 0", err) + } + if body != "quarterly\nyearly\n" { + t.Errorf("body = %q, want the report names over the socket", body) + } + if !path.gone() { + t.Errorf("%s is still there once the program exited, want it removed", path) + } +} + +func TestMainRefusesASocketWithoutTheHeaderSetting(t *testing.T) { + t.Parallel() + + got := runExample(t, "", []string{"MYAPP_SOCKET=" + string(socketAt(t))}, "serve") + + if got.code != gonsole.ExitFailed { + t.Errorf("code = %d, want 1, stderr %q", got.code, got.stderr) + } + if want := "myapp: MYAPP_CLIENT_ADDR_HEADER is required to serve on a socket\n"; got.stderr != want { + t.Errorf("stderr = %q, want %q", got.stderr, want) + } +} diff --git a/gonsole/exec_test.go b/gonsole/exec_test.go index a7aadf4..5cdaa17 100644 --- a/gonsole/exec_test.go +++ b/gonsole/exec_test.go @@ -5,8 +5,8 @@ package gonsole_test import ( "bufio" "bytes" + "context" "errors" - "io" "os" "os/exec" "slices" @@ -35,17 +35,65 @@ func TestMain(m *testing.M) { os.Exit(m.Run()) } -// exampleEnvironment returns the environment of the example program, no inherited MYAPP_ variable and variables added. +// settingEntry reports whether the environment entry sets a MYAPP_ variable or PORT. +func settingEntry(entry string) bool { + return strings.HasPrefix(entry, "MYAPP_") || strings.HasPrefix(entry, "PORT=") +} + +// exampleEnvironment returns the environment of the example program, no inherited setting and variables added. func exampleEnvironment(variables ...string) []string { - inherited := slices.DeleteFunc(os.Environ(), func(entry string) bool { return strings.HasPrefix(entry, "MYAPP_") }) + inherited := slices.DeleteFunc(os.Environ(), settingEntry) return append(append(inherited, exampleSwitch+"=1"), variables...) } +// exampleBound is how long a test lets the example program run before it kills it. +const exampleBound = 10 * time.Second + +// exampleCommand returns the example program over args and extra variables, killed once exampleBound passes. +func exampleCommand(t *testing.T, variables []string, args ...string) *exec.Cmd { + t.Helper() + ctx, cancel := context.WithTimeout(t.Context(), exampleBound) + t.Cleanup(cancel) + cmd := exec.CommandContext(ctx, os.Args[0], args...) + cmd.Env = exampleEnvironment(variables...) + return cmd +} + +// launch starts cmd and returns the lines of its stderr. +func launch(t *testing.T, cmd *exec.Cmd) *bufio.Scanner { + t.Helper() + stderr, err := cmd.StderrPipe() + if err != nil { + t.Fatalf("piping stderr: %v", err) + } + if err := cmd.Start(); err != nil { + t.Fatalf("starting the example program: %v", err) + } + return bufio.NewScanner(stderr) +} + +// untilExit returns every line left in lines. +func untilExit(lines *bufio.Scanner) []string { + var left []string + for lines.Scan() { + left = append(left, lines.Text()) + } + return left +} + +// signalled sends signal to the started cmd and returns the lines it writes to stderr until it exits. +func signalled(t *testing.T, cmd *exec.Cmd, lines *bufio.Scanner, signal os.Signal) []string { + t.Helper() + if err := cmd.Process.Signal(signal); err != nil { + t.Fatalf("signalling: %v", err) + } + return untilExit(lines) +} + // runExample runs the example program in its own process over args, stdin and extra variables and answers its output. func runExample(t *testing.T, stdin string, variables []string, args ...string) result { t.Helper() - cmd := exec.CommandContext(t.Context(), os.Args[0], args...) - cmd.Env = exampleEnvironment(variables...) + cmd := exampleCommand(t, variables, args...) cmd.Stdin = strings.NewReader(stdin) var stdout, stderr bytes.Buffer cmd.Stdout = &stdout @@ -91,8 +139,14 @@ func TestExamplePluginsDescribeTheDemoGroup(t *testing.T) { loaded, err := p.Plugins(t.Context(), gonsole.Call{Describe: true, Env: gonsole.Env{Prefix: "MYAPP_"}}) - if err != nil || loaded.Failed != nil || !slices.Equal(namespaces(loaded.Groups), []string{"demo"}) { - t.Fatalf("Plugins() = %v failing %v, %v, want demo, nil and nil", namespaces(loaded.Groups), loaded.Failed, err) + if err != nil { + t.Fatalf("Plugins() = %v, want nil", err) + } + if loaded.Failed != nil { + t.Fatalf("Plugins() failing %v, want nil", loaded.Failed) + } + if !slices.Equal(namespaces(loaded.Groups), []string{"demo"}) { + t.Fatalf("Plugins() = %v, want demo", namespaces(loaded.Groups)) } if err := p.Check(loaded); err != nil { t.Errorf("Check() = %v, want nil", err) @@ -187,27 +241,13 @@ func TestMainServesUntilASignalEndsTheRun(t *testing.T) { t.Run(signal.String(), func(t *testing.T) { t.Parallel() - cmd := exec.CommandContext(t.Context(), os.Args[0], "serve") - cmd.Env = exampleEnvironment("MYAPP_ADDR=127.0.0.1:0") - stderr, err := cmd.StderrPipe() - if err != nil { - t.Fatalf("piping stderr: %v", err) - } - if err := cmd.Start(); err != nil { - t.Fatalf("starting the example program: %v", err) - } - lines := bufio.NewScanner(stderr) + cmd := exampleCommand(t, []string{"MYAPP_ADDR=127.0.0.1:0"}, "serve") + lines := launch(t, cmd) address := listeningAddress(t, lines) body := get(t, address) - if err := cmd.Process.Signal(signal); err != nil { - t.Fatalf("signalling: %v", err) - } - var rest []string - for lines.Scan() { - rest = append(rest, lines.Text()) - } - err = cmd.Wait() + rest := signalled(t, cmd, lines, signal) + err := cmd.Wait() if err != nil { t.Errorf("exit = %v, want 0", err) @@ -222,34 +262,51 @@ func TestMainServesUntilASignalEndsTheRun(t *testing.T) { } } -func TestMainLetsASecondSignalEndACommandThatIgnoresTheFirst(t *testing.T) { +func TestMainEndsAnIdleServeWithExitZero(t *testing.T) { t.Parallel() - cmd := exec.CommandContext(t.Context(), os.Args[0], "report:import", "-yes") - cmd.Env = exampleEnvironment() + got := runExample(t, "", []string{"MYAPP_ADDR=127.0.0.1:0", "MYAPP_IDLE_EXIT=200ms"}, "serve") + + if got.code != gonsole.ExitDone { + t.Errorf("code = %d, want 0 within %v, stderr %q", got.code, exampleBound, got.stderr) + } + idle := strings.Index(got.stderr, `msg="no connection for the idle exit" idle_exit=200ms`) + if down := strings.Index(got.stderr, `msg="shutting down"`); idle < 0 || down < idle { + t.Errorf("stderr = %q, want the idle line before the shutting down line", got.stderr) + } +} + +// importHeldOpen starts report:import -yes with its input held open and returns it and its exit once it reads. +func importHeldOpen(t *testing.T) (*exec.Cmd, <-chan error) { + t.Helper() + cmd := exampleCommand(t, nil, "report:import", "-yes") stdin, err := cmd.StdinPipe() if err != nil { t.Fatalf("piping stdin: %v", err) } - defer func() { _ = stdin.Close() }() - stderr, err := cmd.StderrPipe() - if err != nil { - t.Fatalf("piping stderr: %v", err) - } - if err := cmd.Start(); err != nil { - t.Fatalf("starting the example program: %v", err) - } - stuck := time.AfterFunc(10*time.Second, func() { _ = cmd.Process.Kill() }) - defer stuck.Stop() - lines := bufio.NewScanner(stderr) + t.Cleanup(func() { _ = stdin.Close() }) + lines := launch(t, cmd) if !lines.Scan() || lines.Text() != importing { t.Fatalf("stderr opens with %q, want %q before any signal", lines.Text(), importing) } exited := make(chan error, 1) go func() { - _, _ = io.Copy(io.Discard, stderr) + untilExit(lines) exited <- cmd.Wait() }() + return cmd, exited +} + +// killedBySignal reports whether a signal ended the process state describes. +func killedBySignal(state *os.ProcessState) bool { + status, known := state.Sys().(syscall.WaitStatus) + return known && status.Signaled() +} + +func TestMainLetsASecondSignalEndACommandThatIgnoresTheFirst(t *testing.T) { + t.Parallel() + + cmd, exited := importHeldOpen(t) _ = cmd.Process.Signal(os.Interrupt) select { @@ -261,8 +318,7 @@ func TestMainLetsASecondSignalEndACommandThatIgnoresTheFirst(t *testing.T) { select { case <-exited: - status, known := cmd.ProcessState.Sys().(syscall.WaitStatus) - if !known || !status.Signaled() { + if !killedBySignal(cmd.ProcessState) { t.Errorf("state = %v, want the second signal to end the program", cmd.ProcessState) } case <-time.After(5 * time.Second): diff --git a/gonsole/idle.go b/gonsole/idle.go new file mode 100644 index 0000000..aae4a1b --- /dev/null +++ b/gonsole/idle.go @@ -0,0 +1,157 @@ +// SPDX-License-Identifier: Apache-2.0 + +package gonsole + +import ( + "context" + "io" + "log/slog" + "net" + "net/http" + "sync" + "time" +) + +// idler counts what is open on one server and ends its run once nothing has been open for its window. +type idler struct { + mu sync.Mutex + open int + quiet time.Time + window time.Duration + run context.Context + end context.CancelFunc + logger *slog.Logger + timer *time.Timer +} + +// newIdler returns an idler that calls end once nothing has been open for window, unless run already ended. +func newIdler(run context.Context, end context.CancelFunc, window time.Duration, logger *slog.Logger) *idler { + idle := &idler{quiet: time.Now(), window: window, run: run, end: end, logger: logger} + idle.timer = time.AfterFunc(window, idle.fire) + return idle +} + +// counting returns listener with each connection it accepts counted open until that connection's first close. +func (i *idler) counting(listener net.Listener) net.Listener { + return countedListener{Listener: listener, idle: i} +} + +// wrap returns next, counting each request while it runs. +func (i *idler) wrap(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + i.hold() + defer i.release() + next.ServeHTTP(w, r) + }) +} + +// hold counts one more open thing. +func (i *idler) hold() { + i.mu.Lock() + defer i.mu.Unlock() + i.open++ +} + +// release counts one open thing fewer, starting the window again once nothing is open. +func (i *idler) release() { + i.mu.Lock() + defer i.mu.Unlock() + i.open-- + if i.open == 0 { + i.quiet = time.Now() + i.timer.Reset(i.window) + } +} + +// fire ends the run once nothing has been open for a full window and the run is still live. +func (i *idler) fire() { + i.mu.Lock() + defer i.mu.Unlock() + if !i.due() || i.run.Err() != nil { + return + } + i.logger.Info("no connection for the idle exit", "idle_exit", i.window) + i.end() +} + +// due reports whether nothing is open and nothing has been for a full window. +func (i *idler) due() bool { + return i.open == 0 && time.Since(i.quiet) >= i.window +} + +// stop stops the timer and ends the run. +func (i *idler) stop() { + i.timer.Stop() + i.end() +} + +// idling returns the run of srv under ctx and listener counted for l's idle exit, zero never, with its stop. +func (l Listening) idling( + ctx context.Context, srv *http.Server, listener net.Listener, logger *slog.Logger, +) (context.Context, net.Listener, func()) { + if l.IdleExit == 0 { + return ctx, listener, func() {} + } + run, end := context.WithCancel(ctx) + idle := newIdler(run, end, l.IdleExit, logger) + srv.Handler = idle.wrap(srv.Handler) + return run, idle.counting(listener), idle.stop +} + +// countedListener is a listener whose idler counts each connection it accepts until that connection's first close. +type countedListener struct { + net.Listener + idle *idler +} + +// Accept waits for the next connection and counts it open until its first close. +func (c countedListener) Accept() (net.Conn, error) { + conn, err := c.Listener.Accept() + if err != nil { + return nil, err + } + c.idle.hold() + return countedOnce(conn, sync.OnceFunc(c.idle.release)), nil +} + +// closeWriter is a connection whose writing side closes alone. +type closeWriter interface { + CloseWrite() error +} + +// countedOnce returns conn calling closed when it closes, keeping its half close when it has one. +func countedOnce(conn net.Conn, closed func()) net.Conn { + counted := countedConn{Conn: conn, closed: closed} + if half, ok := conn.(closeWriter); ok { + return halfClosingConn{countedConn: counted, half: half} + } + return counted +} + +// countedConn is a connection that calls closed each time it closes. +type countedConn struct { + net.Conn + closed func() +} + +// Close closes the connection and calls closed. +func (c countedConn) Close() error { + defer c.closed() + return c.Conn.Close() +} + +// ReadFrom copies r into the connection as io.Copy does, keeping the connection's own sendfile path. +func (c countedConn) ReadFrom(r io.Reader) (int64, error) { + return io.Copy(c.Conn, r) +} + +// halfClosingConn is a counted connection that keeps the half close of the connection it counts. +type halfClosingConn struct { + countedConn + half closeWriter +} + +// CloseWrite closes the writing side of the connection. +func (c halfClosingConn) CloseWrite() error { + return c.half.CloseWrite() +} diff --git a/gonsole/internal/exampleapp/program.go b/gonsole/internal/exampleapp/program.go index 964a734..b856385 100644 --- a/gonsole/internal/exampleapp/program.go +++ b/gonsole/internal/exampleapp/program.go @@ -5,7 +5,6 @@ package exampleapp import ( "bufio" - "cmp" "context" "errors" "fmt" @@ -80,8 +79,12 @@ func serve(ctx context.Context, call gonsole.Call) error { if err != nil { return err } - srv := gonsole.NewServer(cmp.Or(call.Env.Value("ADDR"), "localhost:8080"), http.HandlerFunc(answer), timeouts) - return gonsole.Serve(ctx, srv, timeouts, nil, slog.New(slog.NewTextHandler(call.Stderr, nil))) + listening, err := call.Env.Listening(gonsole.Listening{Address: "localhost:8080"}) + if err != nil { + return err + } + srv := gonsole.NewServer("", http.HandlerFunc(answer), timeouts) + return gonsole.ServeListening(ctx, srv, timeouts, listening, nil, slog.New(slog.NewTextHandler(call.Stderr, nil))) } // answer writes the names of the reports. diff --git a/gonsole/listen.go b/gonsole/listen.go new file mode 100644 index 0000000..c766a62 --- /dev/null +++ b/gonsole/listen.go @@ -0,0 +1,158 @@ +// SPDX-License-Identifier: Apache-2.0 + +package gonsole + +import ( + "cmp" + "context" + "errors" + "fmt" + "log/slog" + "net" + "net/http" + "slices" + "strconv" + "time" +) + +// highestPort is the highest port TCP names. +const highestPort = 65535 + +// errIdleUnbounded is the refusal of an idle exit while a connection to the server can wait forever. +var errIdleUnbounded = errors.New( + "gonsole: Listening.IdleExit needs the server's ReadTimeout above zero, " + + "and its IdleTimeout and ReadHeaderTimeout not negative") + +// Listening is where a server listens, whose visitor address it believes and when it stops for want of work. +type Listening struct { + // Network is unix for a socket file or tcp for an address, empty meaning tcp. + Network string + // Address is the socket's path, the SOCKET setting, or the TCP host and port, from ADDR or PORT. + Address string + // ClientAddrHeader names the header holding the visitor's address on a socket, the CLIENT_ADDR_HEADER setting. + ClientAddrHeader string + // IdleExit is how long the server waits with nothing open before the run ends, zero never, the IDLE_EXIT setting. + IdleExit time.Duration +} + +// Listening returns the listener settings, SOCKET before ADDR before PORT before fallback. +func (e Env) Listening(fallback Listening) (Listening, error) { + var read Listening + var failed [3]error + read.Network, read.Address, failed[0] = e.where(fallback) + read.ClientAddrHeader = cmp.Or(e.Value("CLIENT_ADDR_HEADER"), fallback.ClientAddrHeader) + failed[1] = e.needsHeader(read) + read.IdleExit, failed[2] = e.Duration("IDLE_EXIT", fallback.IdleExit, AllowZero()) + if err := errors.Join(failed[:]...); err != nil { + return Listening{}, err + } + return read, nil +} + +// where returns the network and address the first set of SOCKET, ADDR and PORT names, else the fallback's. +func (e Env) where(fallback Listening) (string, string, error) { + if path := e.Value("SOCKET"); path != "" { + return e.socket(path) + } + if address := e.Value("ADDR"); address != "" { + return "tcp", address, nil + } + port, err := Env{Getenv: e.Getenv}.Count("PORT", 0, AtMost(highestPort)) + if err != nil { + return "", "", err + } + if port > 0 { + return "tcp", ":" + strconv.Itoa(port), nil + } + return fallback.Network, fallback.Address, nil +} + +// socket returns the socket at path, an error naming the SOCKET setting when this system cannot bind it. +func (e Env) socket(path string) (string, string, error) { + if err := checkSocketPath(path); err != nil { + return "", "", fmt.Errorf("%s: %w", e.Key("SOCKET"), err) + } + return "unix", path, nil +} + +// needsHeader returns an error naming the CLIENT_ADDR_HEADER setting when l serves on a socket without a header. +func (e Env) needsHeader(l Listening) error { + if l.headless() { + return fmt.Errorf("%s is required to serve on a socket", e.Key("CLIENT_ADDR_HEADER")) + } + return nil +} + +// ServeListening is Serve on the listener l names in place of srv.Addr, ending the run once l's idle exit passes. +func ServeListening( + ctx context.Context, srv *http.Server, t Timeouts, l Listening, stop func(context.Context) error, + logger *slog.Logger, +) error { + if err := errors.Join(refuseGraces(t), refuseListening(l), refuseIdle(l, srv)); err != nil { + return err + } + listener, err := listenOn(ctx, l) + if err != nil { + return errors.Join(fmt.Errorf("http server: %w", err), stopWithin(ctx, t, stop)) + } + logger = cmp.Or(logger, slog.New(slog.DiscardHandler)) + srv.Handler = visitors(l.ClientAddrHeader, cmp.Or[http.Handler](srv.Handler, http.DefaultServeMux), logger) + run, counted, stopIdling := l.idling(ctx, srv, listener, logger) + defer stopIdling() + return serveOn(run, srv, counted, t, stop, logger) +} + +// refuseListening returns an error naming each field of l that ServeListening cannot serve on. +func refuseListening(l Listening) error { + var refused []error + if !slices.Contains([]string{"", "tcp", "unix"}, l.Network) { + refused = append(refused, fmt.Errorf("gonsole: Listening.Network must be unix, tcp or empty, got %q", l.Network)) + } + if l.Address == "" { + refused = append(refused, errors.New("gonsole: Listening.Address must not be empty")) + } + if l.headless() { + refused = append(refused, errors.New("gonsole: Listening.ClientAddrHeader must name a header for a socket")) + } + if err := l.socketPath(); err != nil { + refused = append(refused, fmt.Errorf("gonsole: Listening.Address: %w", err)) + } + return errors.Join(refused...) +} + +// refuseIdle refuses an idle exit of l below zero, or above zero while a connection to srv can wait forever. +func refuseIdle(l Listening, srv *http.Server) error { + switch { + case l.IdleExit < 0: + return fmt.Errorf("gonsole: Listening.IdleExit must not be negative, got %v", l.IdleExit) + case l.IdleExit > 0 && waitsForever(srv): + return errIdleUnbounded + } + return nil +} + +// waitsForever reports whether no timeout of srv bounds a connection's wait for its next request, header or body. +func waitsForever(srv *http.Server) bool { + return srv.ReadTimeout <= 0 || min(srv.IdleTimeout, srv.ReadHeaderTimeout) < 0 +} + +// listenOn returns a listener on the network and address l names. +func listenOn(ctx context.Context, l Listening) (net.Listener, error) { + if l.Network == "unix" { + return listenSocket(ctx, l.Address) + } + return net.Listen("tcp", l.Address) +} + +// headless reports whether l serves on a socket with no header naming the visitor's address. +func (l Listening) headless() bool { + return l.Network == "unix" && l.ClientAddrHeader == "" +} + +// socketPath refuses the path of a socket listener l when this system cannot bind it, nothing for TCP. +func (l Listening) socketPath() error { + if l.Network != "unix" { + return nil + } + return checkSocketPath(l.Address) +} diff --git a/gonsole/listen_internal_test.go b/gonsole/listen_internal_test.go new file mode 100644 index 0000000..0f684b2 --- /dev/null +++ b/gonsole/listen_internal_test.go @@ -0,0 +1,601 @@ +// SPDX-License-Identifier: Apache-2.0 + +package gonsole + +import ( + "bytes" + "context" + "errors" + "io" + "log/slog" + "net" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "reflect" + "strings" + "sync" + "sync/atomic" + "testing" + "testing/synctest" + "time" +) + +func TestVisitorAddrReadsOneAddress(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + values []string + want string + ok bool + }{ + {"an IPv4 address", []string{"203.0.113.7"}, "203.0.113.7:0", true}, + {"an IPv4 address with its port", []string{"203.0.113.7:8443"}, "203.0.113.7:8443", true}, + {"an IPv6 address", []string{"2001:db8::1"}, "[2001:db8::1]:0", true}, + {"an IPv6 address with its port", []string{"[2001:db8::1]:443"}, "[2001:db8::1]:443", true}, + {"an IPv4-mapped address", []string{"::ffff:203.0.113.7"}, "203.0.113.7:0", true}, + {"an address with a zone", []string{"fe80::1%eth0"}, "[fe80::1]:0", true}, + {"no value", nil, "", false}, + {"an empty value", []string{""}, "", false}, + {"two values", []string{"203.0.113.7", "198.51.100.2"}, "", false}, + {"a comma list", []string{"203.0.113.7, 198.51.100.2"}, "", false}, + {"a comma list after a zone", []string{"fe80::1%eth0, 203.0.113.7"}, "", false}, + {"a comma list after a zone with a port", []string{"[fe80::1%eth0,203.0.113.7]:443"}, "", false}, + {"a word", []string{"unknown"}, "", false}, + {"a port above the highest", []string{"203.0.113.7:99999"}, "", false}, + {"an IPv6 address in brackets with no port", []string{"[2001:db8::1]"}, "", false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + got, ok := visitorAddr(tc.values) + + if got != tc.want || ok != tc.ok { + t.Errorf("visitorAddr(%q) = %q, %t, want %q, %t", tc.values, got, ok, tc.want, tc.ok) + } + }) + } +} + +// pipeAddr is the address of every end of an in-memory pipe. +type pipeAddr struct{} + +// Network names the pipe. +func (pipeAddr) Network() string { return "pipe" } + +// String names the pipe. +func (pipeAddr) String() string { return "pipe" } + +// pipeListener hands out the server ends of the in-memory pipes its dial makes. +type pipeListener struct { + conns chan net.Conn + closed chan struct{} + once sync.Once +} + +// newPipeListener returns a pipe listener that holds up to four connections nobody accepted yet. +func newPipeListener() *pipeListener { + return &pipeListener{conns: make(chan net.Conn, 4), closed: make(chan struct{})} +} + +// Accept returns the server end of the next pipe, or net.ErrClosed once the listener closed. +func (l *pipeListener) Accept() (net.Conn, error) { + select { + case conn := <-l.conns: + return conn, nil + case <-l.closed: + return nil, net.ErrClosed + } +} + +// Close makes every later Accept fail. +func (l *pipeListener) Close() error { + l.once.Do(func() { close(l.closed) }) + return nil +} + +// Addr returns the pipe's address. +func (l *pipeListener) Addr() net.Addr { + return pipeAddr{} +} + +// dial returns the client end of a new pipe whose server end the listener hands out. +func (l *pipeListener) dial() net.Conn { + client, server := net.Pipe() + l.conns <- server + return client +} + +// idleWindow is the idle exit the idler tests run under, in the fake time of a synctest bubble. +const idleWindow = 10 * time.Second + +// tick is the shortest step of fake time. +const tick = time.Nanosecond + +// ledger is a log target that keeps every line, safe to read while another goroutine writes. +type ledger struct { + mu sync.Mutex + text strings.Builder +} + +// Write keeps p. +func (l *ledger) Write(p []byte) (int, error) { + l.mu.Lock() + defer l.mu.Unlock() + return l.text.Write(p) +} + +// String returns everything kept. +func (l *ledger) String() string { + l.mu.Lock() + defer l.mu.Unlock() + return l.text.String() +} + +// idleRun is an idler under test with the run it ends, how often it ended it and what it logged. +type idleRun struct { + idle *idler + run context.Context + ends *atomic.Int32 + logged *ledger +} + +// startIdle returns an idler over a run of parent that ends once nothing has been open for idleWindow. +func startIdle(parent context.Context) idleRun { + run, end := context.WithCancel(parent) + ends := new(atomic.Int32) + logged := new(ledger) + counted := func() { + ends.Add(1) + end() + } + idle := newIdler(run, counted, idleWindow, slog.New(slog.NewTextHandler(logged, nil))) + return idleRun{idle: idle, run: run, ends: ends, logged: logged} +} + +// saidIdle reports whether the idler logged its idle line. +func (r idleRun) saidIdle() bool { + return strings.Contains(r.logged.String(), `level=INFO msg="no connection for the idle exit" idle_exit=10s`) +} + +// liveAfter lets d of fake time pass and reports whether run is still live. +func liveAfter(run context.Context, d time.Duration) bool { + time.Sleep(d) + synctest.Wait() + return run.Err() == nil +} + +// endsAWindowAfter fails t unless run, from now, is live a tick before a window passes and ended once it has. +func endsAWindowAfter(t *testing.T, run context.Context, event string) { + t.Helper() + closing := liveAfter(run, idleWindow-tick) + closed := liveAfter(run, tick) + if !closing { + t.Errorf("the run ended a tick before a window passed since %s", event) + } + if closed { + t.Errorf("the run is live a window after %s, want it ended", event) + } +} + +// connect returns the server end of a new pipe, accepted through a listener the idler counts. +func (r idleRun) connect(t *testing.T) net.Conn { + t.Helper() + pipes := newPipeListener() + pipes.dial() + conn, err := r.idle.counting(pipes).Accept() + if err != nil { + t.Fatalf("accepting a pipe: %v", err) + } + return conn +} + +// windowed is a listener whose idle exit is idleWindow. +var windowed = Listening{Address: "127.0.0.1:0", IdleExit: idleWindow} + +func TestIdlerEndsTheRunAtTheWindow(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + r := startIdle(t.Context()) + defer r.idle.stop() + + early := liveAfter(r.run, idleWindow-tick) + late := liveAfter(r.run, tick) + + if !early { + t.Errorf("the run ended a tick before the window passed") + } + if late { + t.Errorf("the run is live once the window passed with nothing open, want it ended") + } + if !r.saidIdle() { + t.Errorf("log = %q, want the idle line", r.logged.String()) + } + }) +} + +func TestIdlerWaitsWhileAConnectionIsOpen(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + r := startIdle(t.Context()) + defer r.idle.stop() + + conn := r.connect(t) + open := liveAfter(r.run, 2*idleWindow) + _ = conn.Close() + + if !open { + t.Errorf("the run ended while a connection was open") + } + endsAWindowAfter(t, r.run, "the close") + }) +} + +func TestIdlerCountsAConnectionClosedTwiceOnce(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + r := startIdle(t.Context()) + defer r.idle.stop() + first := r.connect(t) + second := r.connect(t) + + _ = first.Close() + _ = first.Close() + open := liveAfter(r.run, 2*idleWindow) + _ = second.Close() + + if !open { + t.Errorf("the run ended while a second connection was open, want the first counted closed once") + } + endsAWindowAfter(t, r.run, "the last close") + }) +} + +func TestIdlerStartsTheWindowAgainAfterAQuickConnection(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + r := startIdle(t.Context()) + defer r.idle.stop() + time.Sleep(idleWindow / 2) + + _ = r.connect(t).Close() + first := liveAfter(r.run, idleWindow/2) + closing := liveAfter(r.run, idleWindow/2-tick) + closed := liveAfter(r.run, tick) + + if !first { + t.Errorf("the run ended when the first window passed, want the window started again at the close") + } + if !closing { + t.Errorf("the run ended a tick before a window passed since the close") + } + if closed { + t.Errorf("the run is live a window after the close, want it ended") + } + }) +} + +func TestIdlerWaitsAWindowAfterAStaleFire(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + r := startIdle(t.Context()) + defer r.idle.stop() + conn := r.connect(t) + time.Sleep(idleWindow) + synctest.Wait() + + _ = conn.Close() + r.idle.fire() + + if r.run.Err() != nil { + t.Errorf("the run ended on a timer fire right after the close, want it to wait a window from the close") + } + endsAWindowAfter(t, r.run, "the close") + }) +} + +func TestIdlerWaitsForARunningHandler(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + r := startIdle(t.Context()) + defer r.idle.stop() + release := make(chan struct{}) + handler := r.idle.wrap(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { <-release })) + + conn := r.connect(t) + go handler.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/", nil)) + synctest.Wait() + _ = conn.Close() + running := liveAfter(r.run, 2*idleWindow) + close(release) + + if !running { + t.Errorf("the run ended while a handler ran after its connection closed") + } + endsAWindowAfter(t, r.run, "the handler returned") + }) +} + +// takeOver is a handler that takes its connection over from net/http and hands it to taken. +func takeOver(t *testing.T, taken chan<- net.Conn) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + conn, _, err := http.NewResponseController(w).Hijack() + if err != nil { + t.Errorf("taking the connection over: %v", err) + return + } + taken <- conn + }) +} + +func TestIdlingCountsAConnectionAHandlerTakesOverUntilItCloses(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + taken := make(chan net.Conn, 1) + srv := &http.Server{Handler: takeOver(t, taken)} + pipes := newPipeListener() + run, listener, stop := windowed.idling(t.Context(), srv, pipes, slog.New(slog.DiscardHandler)) + defer stop() + served := make(chan error, 1) + go func() { served <- srv.Serve(listener) }() + client := pipes.dial() + _, _ = io.WriteString(client, "GET /reports HTTP/1.1\r\nHost: reports.example\r\n\r\n") + conn := <-taken + synctest.Wait() + + open := liveAfter(run, 2*idleWindow) + _ = conn.Close() + + if !open { + t.Errorf("the run ended while a connection a handler took over was open after its handler returned") + } + endsAWindowAfter(t, run, "the close of the connection the handler took over") + _ = client.Close() + _ = srv.Close() + <-served + }) +} + +// halfCloser is a connection whose writing side closes alone. +type halfCloser interface { + CloseWrite() error +} + +// reportFile returns a file in a new folder of t that holds 136 KB of report names. +func reportFile(t *testing.T) (*os.File, []byte) { + t.Helper() + body := bytes.Repeat([]byte("quarterly\n"), 13600) + path := filepath.Join(t.TempDir(), "reports.txt") + if err := os.WriteFile(path, body, 0o600); err != nil { + t.Fatalf("writing %s: %v", path, err) + } + file, err := os.Open(path) + if err != nil { + t.Fatalf("opening %s: %v", path, err) + } + t.Cleanup(func() { _ = file.Close() }) + return file, body +} + +// acceptTCP returns both ends of a new loopback TCP connection, the server's end through a listener r counts. +func (r idleRun) acceptTCP(t *testing.T) (client, counted net.Conn) { + t.Helper() + tcp, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listening: %v", err) + } + t.Cleanup(func() { _ = tcp.Close() }) + client, err = net.Dial("tcp", tcp.Addr().String()) + if err != nil { + t.Fatalf("dialling %s: %v", tcp.Addr(), err) + } + t.Cleanup(func() { _ = client.Close() }) + counted, err = r.idle.counting(tcp).Accept() + if err != nil { + t.Fatalf("accepting: %v", err) + } + t.Cleanup(func() { _ = counted.Close() }) + return client, counted +} + +func TestCountedConnectionsKeepTheFastPathsOfTCP(t *testing.T) { + t.Parallel() + + r := startIdle(t.Context()) + defer r.idle.stop() + client, conn := r.acceptTCP(t) + file, body := reportFile(t) + half, halves := conn.(halfCloser) + from, copies := conn.(io.ReaderFrom) + if !halves || !copies { + t.Fatalf("the counted TCP connection keeps CloseWrite %t and ReadFrom %t, want both kept", halves, copies) + } + received := make(chan []byte, 1) + go func() { + got, _ := io.ReadAll(client) + received <- got + }() + + sent, sendErr := from.ReadFrom(file) + closeErr := half.CloseWrite() + + if got := <-received; !bytes.Equal(got, body) { + t.Errorf("the client read %d bytes before the half close, want all %d", len(got), len(body)) + } + if sent != int64(len(body)) || sendErr != nil { + t.Errorf("ReadFrom() = %d, %v, want %d and nil", sent, sendErr, len(body)) + } + if closeErr != nil { + t.Errorf("CloseWrite() = %v, want nil", closeErr) + } +} + +func TestCountedConnectionsInventNoHalfCloseOnAPipe(t *testing.T) { + t.Parallel() + + r := startIdle(t.Context()) + defer r.idle.stop() + conn := r.connect(t) + defer func() { _ = conn.Close() }() + + if _, halves := conn.(halfCloser); halves { + t.Errorf("the counted pipe offers CloseWrite, want no half close the pipe lacks") + } +} + +// recovered serves one request through handler and returns what it panicked with. +func recovered(handler http.Handler) (panicked any) { + defer func() { panicked = recover() }() + handler.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/", nil)) + return nil +} + +func TestIdlerReleasesAHandlerThatPanics(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + r := startIdle(t.Context()) + defer r.idle.stop() + handler := r.idle.wrap(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { panic(http.ErrAbortHandler) })) + time.Sleep(idleWindow / 2) + + panicked := recovered(handler) + + if panicked != http.ErrAbortHandler { + t.Errorf("the wrapped handler panicked with %v, want the handler's own panic passed on", panicked) + } + endsAWindowAfter(t, r.run, "the handler panicked") + }) +} + +func TestIdlerStaysQuietAfterTheRunEnds(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + parent, endParent := context.WithCancel(t.Context()) + r := startIdle(parent) + defer r.idle.stop() + + endParent() + time.Sleep(idleWindow) + synctest.Wait() + + if r.saidIdle() { + t.Errorf("log = %q, want no idle line once the run ended", r.logged.String()) + } + if ends := r.ends.Load(); ends != 0 { + t.Errorf("the idler ended the run %d times after it ended, want none", ends) + } + }) +} + +// sameHook reports whether got is the hook want, the one function. +func sameHook(got, want func(net.Conn, http.ConnState)) bool { + return reflect.ValueOf(got).Pointer() == reflect.ValueOf(want).Pointer() +} + +func TestIdlingLeavesTheServersOwnConnState(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + own := func(net.Conn, http.ConnState) {} + srv := &http.Server{ConnState: own} + + _, _, stop := windowed.idling(t.Context(), srv, newPipeListener(), slog.New(slog.DiscardHandler)) + stop() + + if !sameHook(srv.ConnState, own) { + t.Errorf("ConnState changed, want the server's own hook left in place") + } + }) +} + +func TestIdlingCountsTheServersRunningRequests(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + release := make(chan struct{}) + srv := &http.Server{Handler: http.HandlerFunc(func(http.ResponseWriter, *http.Request) { <-release })} + run, _, stop := windowed.idling(t.Context(), srv, newPipeListener(), slog.New(slog.DiscardHandler)) + defer stop() + + go srv.Handler.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/", nil)) + running := liveAfter(run, 2*idleWindow) + close(release) + returned := liveAfter(run, idleWindow) + + if !running { + t.Errorf("the run ended while a request of the server ran, want the idler to count it") + } + if returned { + t.Errorf("the run is live a window after the request returned, want it ended") + } + }) +} + +func TestServeListeningStopsItsIdlerWhenServingFails(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + timeouts := Timeouts{ReadHeader: time.Second, Read: time.Second, Idle: time.Minute, Grace: time.Second, + CancelGrace: time.Second, StopGrace: time.Second} + srv := NewServer("", http.NotFoundHandler(), timeouts) + _ = srv.Close() + logged := new(ledger) + at := Listening{Address: "127.0.0.1:0", IdleExit: idleWindow} + + err := ServeListening(t.Context(), srv, timeouts, at, nil, slog.New(slog.NewTextHandler(logged, nil))) + time.Sleep(idleWindow) + synctest.Wait() + + if !errors.Is(err, http.ErrServerClosed) { + t.Errorf("ServeListening() = %v, want the serving failure", err) + } + if strings.Contains(logged.String(), "no connection for the idle exit") { + t.Errorf("log = %q, want no idle line once ServeListening returned", logged.String()) + } + }) +} + +// still is a handler that answers nothing, comparable with ==. +type still struct{} + +// ServeHTTP answers nothing. +func (still) ServeHTTP(http.ResponseWriter, *http.Request) {} + +func TestIdlingWithZeroLeavesTheServerAlone(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + own := func(net.Conn, http.ConnState) {} + srv := &http.Server{ConnState: own, Handler: still{}} + pipes := newPipeListener() + never := Listening{Address: "127.0.0.1:0"} + + run, listener, stop := never.idling(t.Context(), srv, pipes, slog.New(slog.DiscardHandler)) + stop() + + if run != t.Context() { + t.Errorf("idling() returned a new context, want the one it was given") + } + if listener != net.Listener(pipes) { + t.Errorf("idling() returned the listener %T, want the one it was given", listener) + } + if !sameHook(srv.ConnState, own) { + t.Errorf("ConnState changed, want the server's own hook left in place") + } + if srv.Handler != http.Handler(still{}) { + t.Errorf("Handler = %T, want the server's own handler left in place", srv.Handler) + } + }) +} diff --git a/gonsole/listen_test.go b/gonsole/listen_test.go new file mode 100644 index 0000000..201db86 --- /dev/null +++ b/gonsole/listen_test.go @@ -0,0 +1,593 @@ +// SPDX-License-Identifier: Apache-2.0 + +package gonsole_test + +import ( + "context" + "errors" + "io" + "net" + "net/http" + "net/http/httptest" + "slices" + "strings" + "testing" + "time" + + "github.com/gopherium/framework/gonsole" +) + +// local is the listener a program falls back to when its listener settings are empty. +var local = gonsole.Listening{Address: "localhost:8080"} + +func TestEnvReadsWhereToListen(t *testing.T) { + t.Parallel() + + idling := gonsole.Listening{Address: "localhost:8080", IdleExit: 10 * time.Minute} + withHeader := gonsole.Listening{Address: "localhost:8080", ClientAddrHeader: "X-Client-Addr"} + cases := []struct { + name string + values map[string]string + fallback gonsole.Listening + want gonsole.Listening + err string + }{ + {"no settings", nil, local, local, ""}, + {"an address", map[string]string{"MYAPP_ADDR": "127.0.0.1:9000"}, local, + gonsole.Listening{Network: "tcp", Address: "127.0.0.1:9000"}, ""}, + {"a port alone", map[string]string{"PORT": "8080"}, local, + gonsole.Listening{Network: "tcp", Address: ":8080"}, ""}, + {"an address beats a port", map[string]string{"MYAPP_ADDR": "127.0.0.1:9000", "PORT": "8080"}, local, + gonsole.Listening{Network: "tcp", Address: "127.0.0.1:9000"}, ""}, + {"an address beside a port that is no number", map[string]string{"MYAPP_ADDR": "127.0.0.1:9000", "PORT": "abc"}, + local, gonsole.Listening{Network: "tcp", Address: "127.0.0.1:9000"}, ""}, + {"a zero port", map[string]string{"PORT": "0"}, local, gonsole.Listening{}, + `PORT: must stand above zero, got "0"`}, + {"a negative port", map[string]string{"PORT": "-1"}, local, gonsole.Listening{}, + `PORT: must stand above zero, got "-1"`}, + {"a port above the highest", map[string]string{"PORT": "65536"}, local, gonsole.Listening{}, + `PORT: must stand at or below 65535, got "65536"`}, + {"a port that is a word", map[string]string{"PORT": "abc"}, local, gonsole.Listening{}, + `PORT: must be a whole number, got "abc"`}, + {"a port that is a pipe path", map[string]string{"PORT": `\\.\pipe\myapp`}, local, gonsole.Listening{}, + `PORT: must be a whole number, got "\\\\.\\pipe\\myapp"`}, + {"a header", map[string]string{"MYAPP_CLIENT_ADDR_HEADER": "X-Client-Addr"}, local, withHeader, ""}, + {"a header from the fallback", nil, withHeader, withHeader, ""}, + {"a header setting over the fallback's", map[string]string{"MYAPP_CLIENT_ADDR_HEADER": "X-Real-IP"}, withHeader, + gonsole.Listening{Address: "localhost:8080", ClientAddrHeader: "X-Real-IP"}, ""}, + {"an idle exit from the fallback", nil, idling, idling, ""}, + {"an idle exit", map[string]string{"MYAPP_IDLE_EXIT": "10m"}, local, + gonsole.Listening{Address: "localhost:8080", IdleExit: 10 * time.Minute}, ""}, + {"an idle exit of zero over a fallback above it", map[string]string{"MYAPP_IDLE_EXIT": "0s"}, idling, + local, ""}, + {"an idle exit in bare seconds", map[string]string{"MYAPP_IDLE_EXIT": "600"}, local, gonsole.Listening{}, + `MYAPP_IDLE_EXIT: must be a duration like 30s, got "600"`}, + {"a negative idle exit", map[string]string{"MYAPP_IDLE_EXIT": "-1s"}, local, gonsole.Listening{}, + `MYAPP_IDLE_EXIT: must not be negative, got "-1s"`}, + {"two failures", map[string]string{"PORT": "abc", "MYAPP_IDLE_EXIT": "600"}, local, gonsole.Listening{}, + "PORT: must be a whole number, got \"abc\"\nMYAPP_IDLE_EXIT: must be a duration like 30s, got \"600\""}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + got, err := settings(tc.values).Listening(tc.fallback) + + if got != tc.want || errorText(err) != tc.err { + t.Errorf("Listening() = %+v, %q, want %+v, %q", got, errorText(err), tc.want, tc.err) + } + }) + } +} + +// loopback is a listener on a free loopback port. +var loopback = gonsole.Listening{Address: "127.0.0.1:0"} + +// listeningRun is one ServeListening run a test started, with the address it listens on, its log and its stop calls. +type listeningRun struct { + address string + journal *journal + stopped *stopper + end func() error +} + +// serveListening starts ServeListening over handler on l under timeouts and returns the run once it listens. +func serveListening(t *testing.T, l gonsole.Listening, handler http.Handler, timeouts gonsole.Timeouts) listeningRun { + t.Helper() + ctx, cancel := context.WithCancel(t.Context()) + j := newJournal() + s := &stopper{} + srv := gonsole.NewServer("", handler, timeouts) + done := make(chan error, 1) + go func() { done <- gonsole.ServeListening(ctx, srv, timeouts, l, s.stop, j.logger()) }() + select { + case address := <-j.listening: + return listeningRun{address: address, journal: j, stopped: s, end: func() error { + cancel() + return ended(t, done, timeouts) + }} + case err := <-done: + cancel() + t.Fatalf("ServeListening() returned %v before listening", err) + } + return listeningRun{} +} + +// start is one ServeListening call a test makes, with the server, the timeouts beside it and the listener. +type start struct { + srv *http.Server + timeouts gonsole.Timeouts + at gonsole.Listening +} + +// startOf returns the start of a server over the report names built from timeouts, on l. +func startOf(l gonsole.Listening, timeouts gonsole.Timeouts) start { + return start{srv: gonsole.NewServer("", reportNames(), timeouts), timeouts: timeouts, at: l} +} + +// startRefused fails t unless ServeListening on l under timeouts answers want, with nothing bound or stopped. +func startRefused(t *testing.T, l gonsole.Listening, timeouts gonsole.Timeouts, want string) { + t.Helper() + refusedOn(t, startOf(l, timeouts), want) +} + +// refusedOn fails t unless the start s answers want, with nothing bound or stopped. +func refusedOn(t *testing.T, s start, want string) { + t.Helper() + ctx, cancel := context.WithCancel(t.Context()) + cancel() + var stopped stopper + j := newJournal() + + err := gonsole.ServeListening(ctx, s.srv, s.timeouts, s.at, stopped.stop, j.logger()) + + if errorText(err) != want { + t.Errorf("ServeListening() = %q, want %q", errorText(err), want) + } + if j.said("msg=listening") || stopped.calls != 0 { + t.Errorf("listened %t, stop calls %d, want nothing bound or stopped", j.said("msg=listening"), stopped.calls) + } +} + +func TestServeListeningAnswersWhereItIsTold(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + at gonsole.Listening + }{ + {"an address with no network", loopback}, + {"an address on tcp", gonsole.Listening{Network: "tcp", Address: "127.0.0.1:0"}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + run := serveListening(t, tc.at, reportNames(), defaults) + + body := get(t, run.address) + err := run.end() + + if host, _, _ := net.SplitHostPort(run.address); host != "127.0.0.1" { + t.Errorf("listening on %q, want the host 127.0.0.1 that the listener names", run.address) + } + if body != "quarterly\nyearly\n" || err != nil { + t.Errorf("body = %q, ServeListening() = %v, want the report names and nil", body, err) + } + if !run.stopped.ranOnceWithin(defaults.StopGrace) { + t.Errorf("stop calls = %d, live %t, deadline in %v, want one live call within the stop grace", + run.stopped.calls, run.stopped.live, run.stopped.deadline) + } + }) + } +} + +// idleUnbounded is the refusal of an idle exit while a connection to the server can wait forever. +const idleUnbounded = "gonsole: Listening.IdleExit needs the server's ReadTimeout above zero, " + + "and its IdleTimeout and ReadHeaderTimeout not negative" + +func TestServeListeningRefusesBeforeItListens(t *testing.T) { + t.Parallel() + + noGrace := defaults + noGrace.Grace = 0 + noRead := defaults + noRead.Read = 0 + negativeRead := defaults + negativeRead.Read = -time.Second + negativeIdle := defaults + negativeIdle.Idle = -time.Second + negativeHeader := defaults + negativeHeader.ReadHeader = -time.Second + idleExit := gonsole.Listening{Address: "127.0.0.1:0", IdleExit: time.Minute} + udp := gonsole.Listening{Network: "udp", Address: "127.0.0.1:0"} + cases := []struct { + name string + timeouts gonsole.Timeouts + at gonsole.Listening + err string + }{ + {"a zero grace", noGrace, loopback, "gonsole: Timeouts.Grace must stand above zero, got 0s"}, + {"a udp network", defaults, udp, `gonsole: Listening.Network must be unix, tcp or empty, got "udp"`}, + {"an empty address", defaults, gonsole.Listening{}, "gonsole: Listening.Address must not be empty"}, + {"a negative idle exit", defaults, gonsole.Listening{Address: "127.0.0.1:0", IdleExit: -time.Second}, + "gonsole: Listening.IdleExit must not be negative, got -1s"}, + {"an idle exit with no read timeout", noRead, idleExit, idleUnbounded}, + {"an idle exit beside a negative read timeout", negativeRead, idleExit, idleUnbounded}, + {"an idle exit beside a negative idle timeout", negativeIdle, idleExit, idleUnbounded}, + {"an idle exit beside a negative header timeout", negativeHeader, idleExit, idleUnbounded}, + {"a zero grace beside a udp network", noGrace, udp, "gonsole: Timeouts.Grace must stand above zero, got 0s\n" + + `gonsole: Listening.Network must be unix, tcp or empty, got "udp"`}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + startRefused(t, tc.at, tc.timeouts, tc.err) + }) + } +} + +func TestServeListeningStopsWhenTheListenFails(t *testing.T) { + t.Parallel() + + taken, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("taking a port: %v", err) + } + defer func() { _ = taken.Close() }() + s := stopper{fails: errors.New("the reports plugin did not stop")} + at := gonsole.Listening{Address: taken.Addr().String()} + + err = gonsole.ServeListening(t.Context(), gonsole.NewServer("", reportNames(), defaults), defaults, at, s.stop, nil) + + if !strings.HasPrefix(errorText(err), "http server: listen tcp "+taken.Addr().String()) { + t.Errorf("ServeListening() = %v, want the listen failure", err) + } + if !strings.HasSuffix(errorText(err), "\nthe reports plugin did not stop") { + t.Errorf("ServeListening() = %v, want the stop error joined", err) + } + if !s.ranOnceWithin(defaults.StopGrace) { + t.Errorf("stop calls = %d, live %t, deadline in %v, want one live call within the stop grace", + s.calls, s.live, s.deadline) + } +} + +// finishesInsideTheGrace fails t unless a request ServeListening on l serves when the run ends gets its answer. +func finishesInsideTheGrace(t *testing.T, l gonsole.Listening, fetch func(string) (int, error)) { + t.Helper() + arrived := make(chan struct{}) + proceed := make(chan struct{}) + pending := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + close(arrived) + <-proceed + _, _ = io.WriteString(w, "quarterly\n") + }) + run := serveListening(t, l, pending, defaults) + answered := make(chan int, 1) + go func() { + code, _ := fetch(run.address) + answered <- code + }() + <-arrived + ended := make(chan error, 1) + go func() { ended <- run.end() }() + <-run.journal.down + + close(proceed) + + if code := <-answered; code != http.StatusOK { + t.Errorf("answer %d, want the request that arrived before the shutdown answered 200", code) + } + if err := <-ended; err != nil { + t.Errorf("ServeListening() = %v, want nil", err) + } +} + +// cancelledPastTheGrace fails t unless a request ServeListening on l serves past the grace ends with ErrGraceRanOut. +func cancelledPastTheGrace(t *testing.T, l gonsole.Listening, fetch func(string) (int, error)) { + t.Helper() + arrived := make(chan struct{}) + causes := make(chan error, 1) + waiting := http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { + close(arrived) + <-r.Context().Done() + causes <- context.Cause(r.Context()) + }) + timeouts := defaults + timeouts.Grace = 50 * time.Millisecond + run := serveListening(t, l, waiting, timeouts) + go func() { _, _ = fetch(run.address) }() + <-arrived + + err := run.end() + + if err != nil { + t.Errorf("ServeListening() = %v, want nil once the cancelled request ended", err) + } + select { + case cause := <-causes: + if !errors.Is(cause, gonsole.ErrGraceRanOut) { + t.Errorf("cause = %v, want the grace to have run out", cause) + } + default: + t.Errorf("the request was still running when ServeListening returned") + } +} + +func TestServeListeningDrainsAsServeDoes(t *testing.T) { + t.Parallel() + + t.Run("a request inside the grace finishes", func(t *testing.T) { + t.Parallel() + + finishesInsideTheGrace(t, loopback, status) + }) + t.Run("a request past the grace ends with the grace run out", func(t *testing.T) { + t.Parallel() + + cancelledPastTheGrace(t, loopback, status) + }) +} + +// unlogged starts ServeListening over the report names on l with no logger or stop, returning its address and end. +func unlogged(t *testing.T, l gonsole.Listening) (string, func() error) { + t.Helper() + ctx, cancel := context.WithCancel(t.Context()) + srv := gonsole.NewServer("", reportNames(), defaults) + addresses := make(chan string, 1) + srv.BaseContext = func(listener net.Listener) context.Context { + addresses <- listener.Addr().String() + return ctx + } + done := make(chan error, 1) + go func() { done <- gonsole.ServeListening(ctx, srv, defaults, l, nil, nil) }() + select { + case address := <-addresses: + return address, func() error { + cancel() + return ended(t, done, defaults) + } + case err := <-done: + cancel() + t.Fatalf("ServeListening() returned %v before listening", err) + } + return "", nil +} + +func TestServeListeningRunsWithoutALogger(t *testing.T) { + t.Parallel() + + t.Run("serving the report names", func(t *testing.T) { + t.Parallel() + + address, end := unlogged(t, loopback) + + body := get(t, address) + err := end() + + if body != "quarterly\nyearly\n" || err != nil { + t.Errorf("body = %q, ServeListening() = %v, want the report names and nil", body, err) + } + }) + t.Run("ending an idle run", func(t *testing.T) { + t.Parallel() + + at := gonsole.Listening{Address: "127.0.0.1:0", IdleExit: 50 * time.Millisecond} + srv := gonsole.NewServer("", reportNames(), defaults) + done := make(chan error, 1) + go func() { done <- gonsole.ServeListening(t.Context(), srv, defaults, at, nil, nil) }() + + if err := ended(t, done, defaults); err != nil { + t.Errorf("ServeListening() = %v, want nil once the idle exit passed", err) + } + }) +} + +// view is what a handler saw of one request. +type view struct { + remote string + proto string + forwarded string + host string + visitor bool + tls bool + socket bool +} + +// viewer answers the report names and hands what it saw of each request to views. +func viewer(views chan<- view) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, visitor := r.Header["X-Client-Addr"] + views <- view{ + remote: r.RemoteAddr, proto: r.Header.Get("X-Forwarded-Proto"), forwarded: r.Header.Get("X-Forwarded-Host"), + host: r.Host, visitor: visitor, tls: r.TLS != nil, socket: gonsole.OverSocket(r), + } + _, _ = io.WriteString(w, "quarterly\nyearly\n") + }) +} + +// onlyView returns what the handler saw of the one request it served, stopping t when it served none. +func onlyView(t *testing.T, views <-chan view) view { + t.Helper() + select { + case seen := <-views: + return seen + default: + t.Fatalf("the handler served no request") + return view{} + } +} + +// visit returns a GET of the root at address for the reports host, with forwarded headers and values in X-Client-Addr. +func visit(t *testing.T, address string, values []string) *http.Request { + t.Helper() + request, err := http.NewRequestWithContext(t.Context(), http.MethodGet, "http://"+address+"/", nil) + if err != nil { + t.Fatalf("building the request: %v", err) + } + request.Host = "reports.example" + request.Header.Set("X-Forwarded-Proto", "https") + request.Header.Set("X-Forwarded-Host", "reports.example") + request.Header["X-Client-Addr"] = values + return request +} + +// send sends request through client and returns the status code of its answer. +func send(t *testing.T, client *http.Client, request *http.Request) int { + t.Helper() + response, err := client.Do(request) + if err != nil { + t.Fatalf("GET %s: %v", request.URL, err) + } + _ = response.Body.Close() + return response.StatusCode +} + +// believing is a listener on a free loopback port that names X-Client-Addr as the visitor's header. +var believing = gonsole.Listening{Address: "127.0.0.1:0", ClientAddrHeader: "X-Client-Addr"} + +func TestServeListeningDropsTheVisitorHeaderOverTCP(t *testing.T) { + t.Parallel() + + views := make(chan view, 1) + run := serveListening(t, believing, viewer(views), defaults) + + code := send(t, patient, visit(t, run.address, []string{"203.0.113.7"})) + err := run.end() + + if code != http.StatusOK || err != nil { + t.Errorf("answer %d, ServeListening() = %v, want 200 and nil", code, err) + } + seen := onlyView(t, views) + if host, _, _ := net.SplitHostPort(seen.remote); host != "127.0.0.1" { + t.Errorf("RemoteAddr = %q, want the loopback peer the header never replaces over TCP", seen.remote) + } + if seen.visitor { + t.Errorf("the handler saw X-Client-Addr over TCP, want it dropped") + } + if seen.socket { + t.Errorf("OverSocket() = true over TCP, want false") + } +} + +func TestOverSocketIsFalseWithoutAConnection(t *testing.T) { + t.Parallel() + + request := httptest.NewRequest(http.MethodGet, "http://reports.example/", nil) + + if gonsole.OverSocket(request) { + t.Errorf("OverSocket() = true for a request with no connection, want false") + } +} + +func TestServeListeningAnswersThroughTheDefaultMux(t *testing.T) { + t.Parallel() + + run := serveListening(t, believing, nil, defaults) + + code, _ := status(run.address) + err := run.end() + + if code != http.StatusNotFound || err != nil { + t.Errorf("GET answered %d, ServeListening() = %v, want the default mux's 404 and nil", code, err) + } +} + +// lineOf returns the place of the first kept line holding text, -1 when none does. +func (j *journal) lineOf(text string) int { + j.mu.Lock() + defer j.mu.Unlock() + return slices.IndexFunc(j.lines, func(line string) bool { return strings.Contains(line, text) }) +} + +// endsIdle fails t unless ServeListening on l under timeouts ends unprompted with nil, logging the idle exit first. +func endsIdle(t *testing.T, l gonsole.Listening, timeouts gonsole.Timeouts) { + t.Helper() + endsIdleOn(t, startOf(l, timeouts)) +} + +// endsIdleOn fails t unless the start s ends unprompted with nil, logging the idle exit first. +func endsIdleOn(t *testing.T, s start) { + t.Helper() + j := newJournal() + var stopped stopper + done := make(chan error, 1) + go func() { done <- gonsole.ServeListening(t.Context(), s.srv, s.timeouts, s.at, stopped.stop, j.logger()) }() + + err := ended(t, done, s.timeouts) + + if err != nil { + t.Errorf("ServeListening() = %v, want nil once the idle exit passed", err) + } + if !stopped.ranOnceWithin(s.timeouts.StopGrace) { + t.Errorf("stop calls = %d, live %t, deadline in %v, want one live call within the stop grace", + stopped.calls, stopped.live, stopped.deadline) + } + idle := j.lineOf(`level=INFO msg="no connection for the idle exit" idle_exit=` + s.at.IdleExit.String()) + if down := j.lineOf(`msg="shutting down"`); idle < 0 || down < idle { + t.Errorf("log = %q, want the idle line before the shutting down line", j.lines) + } +} + +func TestServeListeningEndsAnIdleRunWithoutAnError(t *testing.T) { + t.Parallel() + + noIdle := defaults + noIdle.Idle = 0 + noHeader := defaults + noHeader.ReadHeader = 0 + cases := []struct { + name string + timeouts gonsole.Timeouts + }{ + {"the default timeouts", defaults}, + {"no idle timeout", noIdle}, + {"no read header timeout", noHeader}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + endsIdle(t, gonsole.Listening{Address: "127.0.0.1:0", IdleExit: 50 * time.Millisecond}, tc.timeouts) + }) + } +} + +// boundedBy returns a server over the report names built from the default timeouts, then changed by change. +func boundedBy(change func(*http.Server)) *http.Server { + srv := gonsole.NewServer("", reportNames(), defaults) + change(srv) + return srv +} + +func TestServeListeningJudgesTheIdleExitOnTheServersTimeouts(t *testing.T) { + t.Parallel() + + idleExit := gonsole.Listening{Address: "127.0.0.1:0", IdleExit: 50 * time.Millisecond} + cases := []struct { + name string + srv *http.Server + }{ + {"a server built with no timeouts", &http.Server{Handler: reportNames()}}, + {"a server whose read timeout was cleared", boundedBy(func(srv *http.Server) { srv.ReadTimeout = 0 })}, + {"a server whose idle timeout is negative", boundedBy(func(srv *http.Server) { srv.IdleTimeout = -time.Second })}, + {"a server whose header timeout is negative", + boundedBy(func(srv *http.Server) { srv.ReadHeaderTimeout = -time.Second })}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + refusedOn(t, start{srv: tc.srv, timeouts: defaults, at: idleExit}, idleUnbounded) + }) + } + t.Run("a bounded server beside timeouts that bound nothing", func(t *testing.T) { + t.Parallel() + + unbounded := defaults + unbounded.ReadHeader, unbounded.Read, unbounded.Idle = 0, 0, 0 + + endsIdleOn(t, start{srv: gonsole.NewServer("", reportNames(), defaults), timeouts: unbounded, at: idleExit}) + }) +} diff --git a/gonsole/serve_test.go b/gonsole/serve_test.go index 9dc7203..df4babf 100644 --- a/gonsole/serve_test.go +++ b/gonsole/serve_test.go @@ -77,9 +77,10 @@ func TestNewServerCarriesTheTimeouts(t *testing.T) { if srv.Addr != "127.0.0.1:8080" { t.Errorf("Addr = %q, want 127.0.0.1:8080", srv.Addr) } - if srv.ReadHeaderTimeout != defaults.ReadHeader || srv.ReadTimeout != defaults.Read || - srv.IdleTimeout != defaults.Idle { - t.Errorf("timeouts = %v, %v, %v, want %+v", srv.ReadHeaderTimeout, srv.ReadTimeout, srv.IdleTimeout, defaults) + carried := gonsole.Timeouts{ReadHeader: srv.ReadHeaderTimeout, Read: srv.ReadTimeout, Idle: srv.IdleTimeout} + want := gonsole.Timeouts{ReadHeader: defaults.ReadHeader, Read: defaults.Read, Idle: defaults.Idle} + if carried != want { + t.Errorf("timeouts = %+v, want %+v", carried, want) } if srv.WriteTimeout != 0 { t.Errorf("WriteTimeout = %v, want none for open streams", srv.WriteTimeout) @@ -291,6 +292,23 @@ func TestServeGivesStopItsOwnGrace(t *testing.T) { } } +// cancelledOnceTheGraceRanOut fails t unless causes holds a request's cancel by the grace, kept for the grace first. +func cancelledOnceTheGraceRanOut(t *testing.T, causes <-chan error, kept *time.Duration, grace time.Duration) { + t.Helper() + select { + case cause := <-causes: + if !errors.Is(cause, gonsole.ErrGraceRanOut) { + t.Errorf("cause = %v, want the grace to have run out", cause) + } + if *kept < grace || *kept >= defaults.CancelGrace { + t.Errorf("request cancelled %v after the shutdown began, want once the grace of %v ran out, well within %v", + *kept, grace, defaults.CancelGrace) + } + default: + t.Errorf("the request was still running when Serve returned") + } +} + func TestServeCancelsARequestThatOutlastsTheGrace(t *testing.T) { t.Parallel() @@ -320,18 +338,7 @@ func TestServeCancelsARequestThatOutlastsTheGrace(t *testing.T) { if err != nil { t.Errorf("Serve() = %v, want nil once the cancelled request ended", err) } - select { - case cause := <-causes: - if !errors.Is(cause, gonsole.ErrGraceRanOut) { - t.Errorf("cause = %v, want the grace to have run out", cause) - } - if kept < grace || kept >= defaults.CancelGrace { - t.Errorf("request cancelled %v after the shutdown began, want once the grace of %v ran out, well within %v", - kept, grace, defaults.CancelGrace) - } - default: - t.Errorf("the request was still running when Serve returned") - } + cancelledOnceTheGraceRanOut(t, causes, &kept, grace) if !endedFirst || !s.ranOnceWithin(defaults.StopGrace) { t.Errorf("request ended before stop %t, stop calls = %d, live %t, deadline in %v, want one live call after it", endedFirst, s.calls, s.live, s.deadline) @@ -371,7 +378,7 @@ func TestServeKeepsARequestsContextLiveThroughTheGrace(t *testing.T) { } } -func TestServeWaitsForAHijackedConnectionBeforeItStops(t *testing.T) { +func TestServeWaitsForAConnectionAHandlerTakesOverBeforeItStops(t *testing.T) { t.Parallel() grace := 50 * time.Millisecond @@ -379,11 +386,11 @@ func TestServeWaitsForAHijackedConnectionBeforeItStops(t *testing.T) { arrived := make(chan struct{}) ended := make(chan struct{}) var kept time.Duration - hijacking := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + takingOver := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { connection, _, err := http.NewResponseController(w).Hijack() close(arrived) if err != nil { - t.Errorf("hijacking the connection: %v", err) + t.Errorf("taking the connection over: %v", err) return } defer func() { _ = connection.Close() }() @@ -400,7 +407,7 @@ func TestServeWaitsForAHijackedConnectionBeforeItStops(t *testing.T) { } return nil } - address, end := serving(t, hijacking, grace, stop, j) + address, end := serving(t, takingOver, grace, stop, j) client, err := net.Dial("tcp", address) if err != nil { t.Fatalf("dialling %s: %v", address, err) @@ -412,10 +419,12 @@ func TestServeWaitsForAHijackedConnectionBeforeItStops(t *testing.T) { err = end() if err != nil || !endedFirst { - t.Errorf("Serve() = %v, hijacked request ended before stop %t, want nil after it ended", err, endedFirst) + t.Errorf("Serve() = %v, the request whose connection was taken over ended before stop %t, want nil after it ended", + err, endedFirst) } if kept < grace { - t.Errorf("hijacked request cancelled %v after the shutdown began, want the whole grace of %v first", kept, grace) + t.Errorf("the request whose connection was taken over was cancelled %v after the shutdown began, "+ + "want the whole grace of %v first", kept, grace) } } @@ -600,10 +609,14 @@ func TestServeKeepsAConnectionsContextLiveThroughTheGrace(t *testing.T) { kept := j.sinceDown() served := ended(t, done, timeouts) - if kept < grace || !errors.Is(context.Cause(conn), gonsole.ErrGraceRanOut) || served != nil { - t.Errorf("connection context ended %v after the shutdown began with %v, Serve() = %v, "+ - "want it live through the grace of %v, then ended by the grace running out, and nil", - kept, context.Cause(conn), served, grace) + if kept < grace { + t.Errorf("connection context ended %v after the shutdown began, want it live through the grace of %v", kept, grace) + } + if cause := context.Cause(conn); !errors.Is(cause, gonsole.ErrGraceRanOut) { + t.Errorf("connection context ended with %v, want it ended by the grace running out", cause) + } + if served != nil { + t.Errorf("Serve() = %v, want nil", served) } } @@ -759,9 +772,14 @@ func TestServeRefusesAGraceThatIsNotAboveZero(t *testing.T) { if taken == nil { _ = again.Close() } - if errorText(err) != tc.err || j.said("msg=listening") || taken != nil || s.calls != 0 { - t.Errorf("Serve() = %q, listened %t, port taken %v, stop calls %d, want %q with nothing bound or stopped", - errorText(err), j.said("msg=listening"), taken, s.calls, tc.err) + if errorText(err) != tc.err { + t.Errorf("Serve() = %q, want %q", errorText(err), tc.err) + } + if j.said("msg=listening") || s.calls != 0 { + t.Errorf("listened %t, stop calls %d, want nothing bound or stopped", j.said("msg=listening"), s.calls) + } + if taken != nil { + t.Errorf("port taken %v, want it left free", taken) } }) } diff --git a/gonsole/socket_internal_linux_test.go b/gonsole/socket_internal_linux_test.go new file mode 100644 index 0000000..9f1463f --- /dev/null +++ b/gonsole/socket_internal_linux_test.go @@ -0,0 +1,136 @@ +// SPDX-License-Identifier: Apache-2.0 + +//go:build linux + +package gonsole + +import ( + "errors" + "os" + "path/filepath" + "syscall" + "testing" +) + +// openLock returns a new open file of the lock file at path, closed once t ends. +func openLock(t *testing.T, path string) *os.File { + t.Helper() + file, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, socketMode) + if err != nil { + t.Fatalf("opening %s: %v", path, err) + } + t.Cleanup(func() { _ = file.Close() }) + return file +} + +// closedLock returns an open file of the lock file at path that is already closed again. +func closedLock(t *testing.T, path string) *os.File { + t.Helper() + file := openLock(t, path) + if err := file.Close(); err != nil { + t.Fatalf("closing %s: %v", path, err) + } + return file +} + +func TestExclusivelyTellsAHeldLockFromAFailure(t *testing.T) { + t.Parallel() + + path := filepath.Join(t.TempDir(), "myapp.sock.lock") + if held, err := exclusively(openLock(t, path)); held || err != nil { + t.Fatalf("exclusively() on a free lock = %t, %v, want false and nil", held, err) + } + cases := []struct { + name string + file *os.File + held bool + err error + }{ + {"a lock another open file holds", openLock(t, path), true, syscall.EWOULDBLOCK}, + {"a file closed before the lock", closedLock(t, path), false, syscall.EBADF}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + held, err := exclusively(tc.file) + + if held != tc.held || !errors.Is(err, tc.err) { + t.Errorf("exclusively() = %t, %v, want %t, %v", held, err, tc.held, tc.err) + } + }) + } +} + +// tryLock takes the lock file at path from an open file of its own and frees it at once. +func tryLock(path string) error { + file, err := os.Open(path) + if err != nil { + return err + } + defer func() { _ = file.Close() }() + return syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB) +} + +// probedListener is a pipe listener that tries the lock file at path as it closes. +type probedListener struct { + *pipeListener + path string + probed error +} + +// Close tries the lock file at path, then makes every later Accept fail. +func (l *probedListener) Close() error { + l.probed = tryLock(l.path) + return l.pipeListener.Close() +} + +func TestLockedHoldsTheLockWhileTheSocketCloses(t *testing.T) { + t.Parallel() + + path := filepath.Join(t.TempDir(), "myapp.sock.lock") + lock := openLock(t, path) + if held, err := exclusively(lock); held || err != nil { + t.Fatalf("exclusively() on a free lock = %t, %v, want false and nil", held, err) + } + listener := &probedListener{pipeListener: newPipeListener(), path: path} + + err := locked{Listener: listener, lock: lock}.Close() + + if !errors.Is(listener.probed, syscall.EWOULDBLOCK) { + t.Errorf("locking %s while the socket closed = %v, want it still held", path, listener.probed) + } + if err != nil { + t.Errorf("Close() = %v, want nil", err) + } +} + +func TestBindLockedRefusesALockItCannotTake(t *testing.T) { + t.Parallel() + + path := filepath.Join(t.TempDir(), "myapp.sock") + socket := socketFile(path) + + listener, err := socket.bindLocked(t.Context(), closedLock(t, socket.lockPath())) + + if want := "lock the socket at " + path + ": bad file descriptor"; err == nil || err.Error() != want { + t.Errorf("bindLocked() = %v, want %q", err, want) + } + if listener != nil { + t.Errorf("bindLocked() listened on %s, want nothing bound", listener.Addr()) + } +} + +func TestClearStaleRefusesAPathItCannotRead(t *testing.T) { + t.Parallel() + + above := filepath.Join(t.TempDir(), "notes") + if err := os.WriteFile(above, []byte("quarterly\n"), 0o600); err != nil { + t.Fatalf("writing %s: %v", above, err) + } + path := filepath.Join(above, "myapp.sock") + + err := socketFile(path).clearStale(t.Context()) + + if want := "lstat " + path + ": not a directory"; err == nil || err.Error() != want { + t.Errorf("clearStale() = %v, want %q", err, want) + } +} diff --git a/gonsole/socket_linux.go b/gonsole/socket_linux.go new file mode 100644 index 0000000..7f3c5f0 --- /dev/null +++ b/gonsole/socket_linux.go @@ -0,0 +1,170 @@ +// SPDX-License-Identifier: Apache-2.0 + +//go:build linux + +package gonsole + +import ( + "context" + "errors" + "fmt" + "io/fs" + "net" + "os" + "strings" + "syscall" +) + +// socketMode is the mode of every socket file ServeListening creates, and of the lock file beside it. +const socketMode = 0o600 + +// socketPathLimit is the longest socket path in bytes the kernel binds. +const socketPathLimit = len(syscall.RawSockaddrUnix{}.Path) - 1 + +// lockSuffix names the lock file beside each socket, the socket's path with .lock added. +const lockSuffix = ".lock" + +// checkSocketPath refuses an abstract socket name and a path longer than the kernel binds. +func checkSocketPath(path string) error { + if strings.HasPrefix(path, "@") { + return complaint("must be a file path, not an abstract name", path) + } + if len(path) > socketPathLimit { + return fmt.Errorf("must be at most %d bytes, got %d", socketPathLimit, len(path)) + } + return nil +} + +// listenSocket returns a listener on a new socket at path that only this account can open, holding its lock file. +func listenSocket(ctx context.Context, path string) (net.Listener, error) { + socket := socketFile(path) + lock, err := os.OpenFile(socket.lockPath(), os.O_RDWR|os.O_CREATE, socketMode) + if err != nil { + return nil, socket.lockFailed(err) + } + listener, err := socket.bindLocked(ctx, lock) + if err != nil { + _ = lock.Close() + return nil, err + } + return locked{Listener: listener, lock: lock}, nil +} + +// exclusively takes the lock on file without waiting, reporting whether another open file holds it. +func exclusively(file *os.File) (bool, error) { + err := syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB) + return errors.Is(err, syscall.EWOULDBLOCK), err +} + +// privately sets the mode of the socket c holds to socketMode before it binds. +func privately(_, _ string, c syscall.RawConn) error { + var fchmod error + cerr := c.Control(func(fd uintptr) { fchmod = syscall.Fchmod(int(fd), socketMode) }) + return errors.Join(cerr, fchmod) +} + +// socketFile is the path of a socket file, with its lock file beside it. +type socketFile string + +// lockPath returns the path of the lock file beside the socket. +func (s socketFile) lockPath() string { + return string(s) + lockSuffix +} + +// bindLocked takes lock without waiting, then clears the socket's path and binds a private socket there. +func (s socketFile) bindLocked(ctx context.Context, lock *os.File) (net.Listener, error) { + held, err := exclusively(lock) + if held { + return nil, s.refuseHeld(ctx) + } + if err != nil { + return nil, s.lockFailed(err) + } + if err := s.clearStale(ctx); err != nil { + return nil, err + } + return s.bindPrivately(ctx) +} + +// lockFailed is the refusal of a start that could not take the lock beside the socket. +func (s socketFile) lockFailed(err error) error { + return fmt.Errorf("lock the socket at %s: %w", s, err) +} + +// refuseHeld names what holds the lock beside the socket, a server that answers on it, else one starting. +func (s socketFile) refuseHeld(ctx context.Context) error { + if s.knock(ctx) == nil { + return s.listensOn() + } + return fmt.Errorf("another server is starting on %s", s) +} + +// listensOn is the refusal of a socket that a server answers on. +func (s socketFile) listensOn() error { + return fmt.Errorf("a server already listens on %s", s) +} + +// knock dials the socket and hangs up, whatever the end of ctx. +func (s socketFile) knock(ctx context.Context) error { + var dialer net.Dialer + conn, err := dialer.DialContext(context.WithoutCancel(ctx), "unix", string(s)) + if err != nil { + return err + } + _ = conn.Close() + return nil +} + +// clearStale removes a socket at the path that nothing listens on, refusing whatever else it finds there. +func (s socketFile) clearStale(ctx context.Context) error { + info, err := os.Lstat(string(s)) + if errors.Is(err, fs.ErrNotExist) { + return nil + } + if err != nil { + return err + } + if info.Mode().Type() != fs.ModeSocket { + return fmt.Errorf("%s is not a socket, so it stays in place", s) + } + err = s.knock(ctx) + if err == nil { + return s.listensOn() + } + if !errors.Is(err, syscall.ECONNREFUSED) { + return fmt.Errorf("check the socket at %s: %w", s, err) + } + return os.Remove(string(s)) +} + +// bindPrivately binds a socket of mode socketMode at the path, removing it again when the umask took bits it needs. +func (s socketFile) bindPrivately(ctx context.Context) (net.Listener, error) { + config := net.ListenConfig{Control: privately} + listener, err := config.Listen(context.WithoutCancel(ctx), "unix", string(s)) + if err != nil { + return nil, err + } + if !s.private() { + _ = listener.Close() + return nil, fmt.Errorf("the socket at %s is not mode %04o, check the umask", s, socketMode) + } + return listener, nil +} + +// private reports whether the path holds a socket of mode socketMode. +func (s socketFile) private() bool { + info, err := os.Lstat(string(s)) + return err == nil && info.Mode() == fs.ModeSocket|socketMode +} + +// locked is a socket listener that frees its lock once its socket file is gone. +type locked struct { + net.Listener + lock *os.File +} + +// Close removes the socket file, then frees the lock. +func (l locked) Close() error { + err := l.Listener.Close() + return errors.Join(err, l.lock.Close()) +} diff --git a/gonsole/socket_linux_test.go b/gonsole/socket_linux_test.go new file mode 100644 index 0000000..8afa844 --- /dev/null +++ b/gonsole/socket_linux_test.go @@ -0,0 +1,525 @@ +// SPDX-License-Identifier: Apache-2.0 + +//go:build linux + +package gonsole_test + +import ( + "context" + "errors" + "fmt" + "io" + "io/fs" + "net" + "net/http" + "os" + "path/filepath" + "strings" + "syscall" + "testing" + "time" + + "github.com/gopherium/framework/gonsole" +) + +// socketPath is a socket path the reader tests name, where nothing ever listens. +const socketPath = "/run/myapp/myapp.sock" + +// pathOfLength returns a socket path of exactly n bytes. +func pathOfLength(n int) string { + return "/tmp/" + strings.Repeat("s", n-len("/tmp/")) +} + +func TestEnvReadsASocketPath(t *testing.T) { + t.Parallel() + + header := "X-Client-Addr" + socket := gonsole.Listening{Network: "unix", Address: socketPath} + headed := gonsole.Listening{Network: "unix", Address: socketPath, ClientAddrHeader: header} + cases := []struct { + name string + values map[string]string + fallback gonsole.Listening + want gonsole.Listening + err string + }{ + {"a socket beats an address and a port", map[string]string{ + "MYAPP_SOCKET": socketPath, "MYAPP_ADDR": "127.0.0.1:9000", "PORT": "8080", "MYAPP_CLIENT_ADDR_HEADER": header, + }, local, gonsole.Listening{Network: "unix", Address: socketPath, ClientAddrHeader: header}, ""}, + {"a socket from the fallback", map[string]string{"MYAPP_CLIENT_ADDR_HEADER": header}, socket, + gonsole.Listening{Network: "unix", Address: socketPath, ClientAddrHeader: header}, ""}, + {"a socket and its header from the fallback", nil, headed, headed, ""}, + {"a fallback socket that an address beats needs no header", map[string]string{"MYAPP_ADDR": "127.0.0.1:9000"}, + socket, gonsole.Listening{Network: "tcp", Address: "127.0.0.1:9000"}, ""}, + {"a fallback socket that only PORT beats", map[string]string{"PORT": "8080", "MYAPP_CLIENT_ADDR_HEADER": header}, + socket, gonsole.Listening{Network: "tcp", Address: ":8080", ClientAddrHeader: header}, ""}, + {"a socket without the header setting", map[string]string{"MYAPP_SOCKET": socketPath}, local, + gonsole.Listening{}, "MYAPP_CLIENT_ADDR_HEADER is required to serve on a socket"}, + {"a path of 107 bytes", map[string]string{"MYAPP_SOCKET": pathOfLength(107), "MYAPP_CLIENT_ADDR_HEADER": header}, + local, gonsole.Listening{Network: "unix", Address: pathOfLength(107), ClientAddrHeader: header}, ""}, + {"a path of 108 bytes", map[string]string{"MYAPP_SOCKET": pathOfLength(108), "MYAPP_CLIENT_ADDR_HEADER": header}, + local, gonsole.Listening{}, "MYAPP_SOCKET: must be at most 107 bytes, got 108"}, + {"an abstract name", map[string]string{"MYAPP_SOCKET": "@myapp", "MYAPP_CLIENT_ADDR_HEADER": header}, local, + gonsole.Listening{}, `MYAPP_SOCKET: must be a file path, not an abstract name, got "@myapp"`}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + got, err := settings(tc.values).Listening(tc.fallback) + + if got != tc.want || errorText(err) != tc.err { + t.Errorf("Listening() = %+v, %q, want %+v, %q", got, errorText(err), tc.want, tc.err) + } + }) + } +} + +func TestServeListeningRefusesASocketBeforeItListens(t *testing.T) { + t.Parallel() + + path := string(socketAt(t)) + cases := []struct { + name string + at gonsole.Listening + err string + }{ + {"a socket with no header", gonsole.Listening{Network: "unix", Address: path}, + "gonsole: Listening.ClientAddrHeader must name a header for a socket"}, + {"a socket path of 108 bytes", + gonsole.Listening{Network: "unix", Address: pathOfLength(108), ClientAddrHeader: "X-Client-Addr"}, + "gonsole: Listening.Address: must be at most 107 bytes, got 108"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + startRefused(t, tc.at, defaults, tc.err) + }) + } +} + +// place is a path on disk that a socket test serves on or puts something at. +type place string + +// shortFolder returns a new folder of a short name under the temporary folder, removed once t ends. +func shortFolder(t *testing.T) string { + t.Helper() + folder, err := os.MkdirTemp("", "gs") + if err != nil { + t.Fatalf("making a folder: %v", err) + } + t.Cleanup(func() { + if err := os.RemoveAll(folder); err != nil { + t.Errorf("removing %s: %v", folder, err) + } + }) + return folder +} + +// socketAt returns the place of a socket in a new short folder of t. +func socketAt(t *testing.T) place { + t.Helper() + return place(filepath.Join(shortFolder(t), "myapp.sock")) +} + +// listening is a listener on the socket at p that reads the visitor's address from X-Client-Addr. +func (p place) listening() gonsole.Listening { + return gonsole.Listening{Network: "unix", Address: string(p), ClientAddrHeader: "X-Client-Addr"} +} + +// client returns a client whose every request dials the socket at p. +func (p place) client() *http.Client { + var dialer net.Dialer + return &http.Client{Timeout: patient.Timeout, Transport: &http.Transport{ + DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { + return dialer.DialContext(ctx, "unix", string(p)) + }, + }} +} + +// answer fetches the root over the socket at p as the program in front does, naming the visitor 203.0.113.7. +func (p place) answer() (*http.Response, error) { + request, err := http.NewRequestWithContext(context.Background(), http.MethodGet, "http://reports.example/", nil) + if err != nil { + return nil, err + } + request.Header.Set("X-Client-Addr", "203.0.113.7") + return p.client().Do(request) +} + +// get fetches the root over the socket at p and returns its body. +func (p place) get(t *testing.T) string { + t.Helper() + response, err := p.answer() + if err != nil { + t.Fatalf("GET over %s: %v", p, err) + } + defer func() { _ = response.Body.Close() }() + body, err := io.ReadAll(response.Body) + if err != nil { + t.Fatalf("reading the answer: %v", err) + } + return string(body) +} + +// status fetches the root over the socket at p and returns the answer's status code, or the error when none came. +func (p place) status() (int, error) { + response, err := p.answer() + if err != nil { + return 0, err + } + _ = response.Body.Close() + return response.StatusCode, nil +} + +// socketStatus is status over the socket at the address a server logged. +func socketStatus(address string) (int, error) { + return place(address).status() +} + +// deadSocket leaves at p a socket file that nothing listens on. +func (p place) deadSocket(t *testing.T) { + t.Helper() + syscall.ForkLock.RLock() + defer syscall.ForkLock.RUnlock() + listener, err := net.ListenUnix("unix", &net.UnixAddr{Name: string(p), Net: "unix"}) + if err != nil { + t.Fatalf("listening on %s: %v", p, err) + } + listener.SetUnlinkOnClose(false) + _ = listener.Close() +} + +// mode returns the type and mode of what is at p, without following a link. +func (p place) mode(t *testing.T) fs.FileMode { + t.Helper() + info, err := os.Lstat(string(p)) + if err != nil { + t.Errorf("reading %s: %v", p, err) + return 0 + } + return info.Mode() +} + +// gone reports whether nothing is at p. +func (p place) gone() bool { + _, err := os.Lstat(string(p)) + return errors.Is(err, fs.ErrNotExist) +} + +// describe returns the type, mode, size and link target of what is at p, or why nothing could be read there. +func (p place) describe() string { + info, err := os.Lstat(string(p)) + if err != nil { + return err.Error() + } + target, _ := os.Readlink(string(p)) + return fmt.Sprintf("%v, %d bytes, linked to %q", info.Mode(), info.Size(), target) +} + +// listenRefused fails t unless ServeListening on l fails its listen with want, serves nothing and calls stop once. +func listenRefused(t *testing.T, l gonsole.Listening, want string) { + t.Helper() + ctx, cancel := context.WithCancel(t.Context()) + cancel() + var s stopper + j := newJournal() + + err := gonsole.ServeListening(ctx, gonsole.NewServer("", reportNames(), defaults), defaults, l, s.stop, j.logger()) + + if errorText(err) != want { + t.Errorf("ServeListening() = %q, want %q", errorText(err), want) + } + if j.said("msg=listening") { + t.Errorf("log = %q, want no listening line", j.lines) + } + if !s.ranOnceWithin(defaults.StopGrace) { + t.Errorf("stop calls = %d, live %t, deadline in %v, want one live call within the stop grace", + s.calls, s.live, s.deadline) + } +} + +// servesPrivately fails t unless ServeListening answers on a socket at path of mode 0600 that is gone once it returns. +func servesPrivately(t *testing.T, path place) { + t.Helper() + run := serveListening(t, path.listening(), reportNames(), defaults) + + body := path.get(t) + mode := path.mode(t) + err := run.end() + + if run.address != string(path) { + t.Errorf("listening on %q, want the socket's path %q", run.address, path) + } + if body != "quarterly\nyearly\n" { + t.Errorf("body = %q, want the report names", body) + } + if err != nil { + t.Errorf("ServeListening() = %v, want nil", err) + } + if mode != fs.ModeSocket|0o600 { + t.Errorf("mode while serving = %v, want a socket of mode 0600", mode) + } + if !path.gone() { + t.Errorf("%s is still there once ServeListening returned, want it removed", path) + } +} + +// refusedForItsMode fails t unless ServeListening at path refuses the socket's mode and leaves no file behind. +func refusedForItsMode(t *testing.T, path place) { + t.Helper() + listenRefused(t, path.listening(), "http server: the socket at "+string(path)+" is not mode 0600, check the umask") + if !path.gone() { + t.Errorf("%s is still there, want the refused socket removed", path) + } +} + +func TestServeListeningServesOnAPrivateSocket(t *testing.T) { + t.Parallel() + + servesPrivately(t, socketAt(t)) +} + +func TestServeListeningMakesTheSocketPrivateUnderAnyUmask(t *testing.T) { + cases := []struct { + name string + umask int + check func(*testing.T, place) + }{ + {"umask 000", 0o000, servesPrivately}, + {"umask 022", 0o022, servesPrivately}, + {"umask 0277", 0o277, refusedForItsMode}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + path := socketAt(t) + restore := syscall.Umask(tc.umask) + t.Cleanup(func() { syscall.Umask(restore) }) + + tc.check(t, path) + }) + } +} + +func TestServeListeningReplacesAStaleSocket(t *testing.T) { + t.Parallel() + + path := socketAt(t) + path.deadSocket(t) + run := serveListening(t, path.listening(), reportNames(), defaults) + + body := path.get(t) + err := run.end() + + if body != "quarterly\nyearly\n" { + t.Errorf("body = %q, want the report names from the server that replaced the stale socket", body) + } + if err != nil { + t.Errorf("ServeListening() = %v, want nil", err) + } +} + +func TestServeListeningRefusesASocketAServerHolds(t *testing.T) { + t.Parallel() + + path := socketAt(t) + first := serveListening(t, path.listening(), reportNames(), defaults) + + listenRefused(t, path.listening(), "http server: a server already listens on "+string(path)) + + body := path.get(t) + if err := first.end(); err != nil { + t.Errorf("the first ServeListening() = %v, want nil", err) + } + if body != "quarterly\nyearly\n" { + t.Errorf("body = %q, want the first server still answering the report names", body) + } +} + +// writeFile puts a regular file at path. +func writeFile(t *testing.T, path place) { + t.Helper() + if err := os.WriteFile(string(path), []byte("quarterly\n"), 0o600); err != nil { + t.Fatalf("writing %s: %v", path, err) + } +} + +// makeFolder puts an empty folder at path. +func makeFolder(t *testing.T, path place) { + t.Helper() + if err := os.Mkdir(string(path), 0o700); err != nil { + t.Fatalf("making %s: %v", path, err) + } +} + +// linkToADeadSocket puts at path a symbolic link to a dead socket in the same folder. +func linkToADeadSocket(t *testing.T, path place) { + t.Helper() + target := place(filepath.Join(filepath.Dir(string(path)), "dead.sock")) + target.deadSocket(t) + if err := os.Symlink(string(target), string(path)); err != nil { + t.Fatalf("linking %s: %v", path, err) + } +} + +// lockedSocket puts at path a dead socket of mode 0o000. +func lockedSocket(t *testing.T, path place) { + t.Helper() + path.deadSocket(t) + if err := os.Chmod(string(path), 0o000); err != nil { + t.Fatalf("locking %s: %v", path, err) + } +} + +// fileAbove puts a regular file where the folder of path should be. +func fileAbove(t *testing.T, path place) { + t.Helper() + writeFile(t, place(filepath.Dir(string(path)))) +} + +func TestServeListeningLeavesWhatIsNotASocket(t *testing.T) { + t.Parallel() + + notASocket := "%[1]s is not a socket, so it stays in place" + cases := []struct { + name string + at string + put func(*testing.T, place) + err string + }{ + {"a regular file", "myapp.sock", writeFile, notASocket}, + {"a folder", "myapp.sock", makeFolder, notASocket}, + {"a symbolic link to a dead socket", "myapp.sock", linkToADeadSocket, notASocket}, + {"a socket this account cannot connect to", "myapp.sock", lockedSocket, + "check the socket at %[1]s: dial unix %[1]s: connect: permission denied"}, + {"a path under a regular file", "notes/myapp.sock", fileAbove, + "lock the socket at %[1]s: open %[1]s.lock: not a directory"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + path := place(filepath.Join(shortFolder(t), tc.at)) + tc.put(t, path) + before := path.describe() + + listenRefused(t, path.listening(), "http server: "+fmt.Sprintf(tc.err, path)) + + if after := path.describe(); after != before { + t.Errorf("%s was %s and is now %s, want it left as it was", path, before, after) + } + }) + } +} + +func TestServeListeningStopsWhenTheSocketsFolderIsMissing(t *testing.T) { + t.Parallel() + + path := place(filepath.Join(shortFolder(t), "missing", "myapp.sock")) + + listenRefused(t, path.listening(), + "http server: lock the socket at "+string(path)+": open "+string(path)+".lock: no such file or directory") +} + +func TestServeListeningDrainsOverASocket(t *testing.T) { + t.Parallel() + + t.Run("a request inside the grace finishes", func(t *testing.T) { + t.Parallel() + + finishesInsideTheGrace(t, socketAt(t).listening(), socketStatus) + }) + t.Run("a request past the grace ends with the grace run out", func(t *testing.T) { + t.Parallel() + + cancelledPastTheGrace(t, socketAt(t).listening(), socketStatus) + }) +} + +func TestServeListeningTakesTheVisitorFromTheHeader(t *testing.T) { + t.Parallel() + + path := socketAt(t) + views := make(chan view, 1) + run := serveListening(t, path.listening(), viewer(views), defaults) + + code := send(t, path.client(), visit(t, "reports.example", []string{"203.0.113.7"})) + err := run.end() + + if code != http.StatusOK || err != nil { + t.Errorf("answer %d, ServeListening() = %v, want 200 and nil", code, err) + } + want := view{ + remote: "203.0.113.7:0", proto: "https", forwarded: "reports.example", host: "reports.example", socket: true, + } + if seen := onlyView(t, views); seen != want { + t.Errorf("the handler saw %+v, want %+v", seen, want) + } +} + +func TestServeListeningRefusesASocketRequestWithoutOneVisitor(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + values []string + }{ + {"no header", nil}, + {"an empty header", []string{""}}, + {"two header lines", []string{"203.0.113.7", "198.51.100.2"}}, + {"a comma list", []string{"fe80::1%eth0, 203.0.113.7"}}, + {"a value that is no address", []string{"unknown"}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + path := socketAt(t) + views := make(chan view, 1) + run := serveListening(t, path.listening(), viewer(views), defaults) + + code := send(t, path.client(), visit(t, "reports.example", tc.values)) + err := run.end() + + if code != http.StatusBadRequest || err != nil { + t.Errorf("answer %d, ServeListening() = %v, want 400 and nil", code, err) + } + if len(views) != 0 { + t.Errorf("the handler served the request, want no handler run") + } + if !run.journal.said(`level=WARN msg="refused a socket request without one visitor address" ` + + "header=X-Client-Addr") { + t.Errorf("log = %q, want the warning naming the header", run.journal.lines) + } + }) + } +} + +func TestServeListeningRefusesASocketRequestWithoutOneVisitorAndNoLogger(t *testing.T) { + t.Parallel() + + path := socketAt(t) + _, end := unlogged(t, path.listening()) + + code := send(t, path.client(), visit(t, "reports.example", nil)) + err := end() + + if code != http.StatusBadRequest || err != nil { + t.Errorf("answer %d, ServeListening() = %v, want 400 and nil", code, err) + } +} + +func TestServeListeningEndsAnIdleRunOnASocket(t *testing.T) { + t.Parallel() + + path := socketAt(t) + at := path.listening() + at.IdleExit = 50 * time.Millisecond + + endsIdle(t, at, defaults) + + if !path.gone() { + t.Errorf("%s is still there once the idle run ended, want it removed", path) + } +} diff --git a/gonsole/socket_lock_linux_test.go b/gonsole/socket_lock_linux_test.go new file mode 100644 index 0000000..9647bb2 --- /dev/null +++ b/gonsole/socket_lock_linux_test.go @@ -0,0 +1,282 @@ +// SPDX-License-Identifier: Apache-2.0 + +//go:build linux + +package gonsole_test + +import ( + "context" + "errors" + "io" + "net" + "net/http" + "net/http/httptrace" + "os" + "path/filepath" + "sync" + "syscall" + "testing" + + "github.com/gopherium/framework/gonsole" +) + +// lockFile returns the place of the lock file beside the socket at p. +func (p place) lockFile() place { + return p + ".lock" +} + +// tryLock takes the lock beside the socket at p from an open file of its own and frees it at once. +func (p place) tryLock() error { + file, err := os.OpenFile(string(p.lockFile()), os.O_RDWR, 0) + if err != nil { + return err + } + defer func() { _ = file.Close() }() + return syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB) +} + +// holdLock takes the lock beside the socket at p from an open file of its own, as another start would, until t ends. +func (p place) holdLock(t *testing.T) { + t.Helper() + file, err := os.OpenFile(string(p.lockFile()), os.O_RDWR|os.O_CREATE, 0o600) + if err != nil { + t.Fatalf("opening %s: %v", p.lockFile(), err) + } + t.Cleanup(func() { _ = file.Close() }) + if err := syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil { + t.Fatalf("locking %s: %v", p.lockFile(), err) + } +} + +// wantLockHeld fails t unless the lock file beside the socket at p is in place and another open file holds it. +func (p place) wantLockHeld(t *testing.T, holder string) { + t.Helper() + if err := p.tryLock(); !errors.Is(err, syscall.EWOULDBLOCK) { + t.Errorf("locking %s once the start was refused = %v, want the lock of %s still held there", + p.lockFile(), err, holder) + } +} + +// noForks keeps the process from starting a child until t ends. +func noForks(t *testing.T) { + t.Helper() + syscall.ForkLock.RLock() + t.Cleanup(syscall.ForkLock.RUnlock) +} + +// boundSocket leaves at p a socket that is bound and never listens, as a start between its bind and its listen does. +func (p place) boundSocket(t *testing.T) { + t.Helper() + fd, err := syscall.Socket(syscall.AF_UNIX, syscall.SOCK_STREAM|syscall.SOCK_CLOEXEC, 0) + if err != nil { + t.Fatalf("making a socket: %v", err) + } + t.Cleanup(func() { _ = syscall.Close(fd) }) + if err := syscall.Bind(fd, &syscall.SockaddrUnix{Name: string(p)}); err != nil { + t.Fatalf("binding %s: %v", p, err) + } +} + +// attempt is one start of ServeListening that a test does not wait for. +type attempt struct { + journal *journal + done chan error + cancel context.CancelFunc +} + +// attemptWith starts ServeListening on srv and l under ctx and returns at once. +func attemptWith(ctx context.Context, srv *http.Server, l gonsole.Listening) attempt { + ctx, cancel := context.WithCancel(ctx) + a := attempt{journal: newJournal(), done: make(chan error, 1), cancel: cancel} + go func() { a.done <- gonsole.ServeListening(ctx, srv, defaults, l, nil, a.journal.logger()) }() + return a +} + +// attemptOn starts ServeListening over the report names on l under ctx and returns at once. +func attemptOn(ctx context.Context, l gonsole.Listening) attempt { + return attemptWith(ctx, gonsole.NewServer("", reportNames(), defaults), l) +} + +// settle waits until the start listens or returns, and reports which with what it returned. +func (a attempt) settle() (bool, error) { + select { + case <-a.journal.listening: + return true, nil + case err := <-a.done: + return false, err + } +} + +// end ends a start that listens and returns what it returned. +func (a attempt) end(t *testing.T) error { + t.Helper() + a.cancel() + return ended(t, a.done, defaults) +} + +func TestServeListeningRefusesAStartThatOverlapsAnotherOnAStaleSocket(t *testing.T) { + t.Parallel() + + path := socketAt(t) + path.deadSocket(t) + var second attempt + var listened bool + var refused error + overlap := sync.OnceFunc(func() { + second = attemptOn(t.Context(), path.listening()) + listened, refused = second.settle() + }) + traced := httptrace.WithClientTrace(t.Context(), &httptrace.ClientTrace{ + ConnectDone: func(string, string, error) { overlap() }, + }) + first := attemptOn(traced, path.listening()) + if serving, err := first.settle(); !serving { + t.Fatalf("the first ServeListening() returned %v before listening", err) + } + + body := path.get(t) + + path.wantLockHeld(t, "the first start") + if listened { + t.Errorf("the second start listened while the first checked the stale socket, want it refused") + _ = second.end(t) + } + if want := "http server: another server is starting on " + string(path); !listened && errorText(refused) != want { + t.Errorf("the second ServeListening() = %q, want %q", errorText(refused), want) + } + if body != "quarterly\nyearly\n" { + t.Errorf("body = %q, want the first server answering the report names", body) + } + if err := first.end(t); err != nil { + t.Errorf("the first ServeListening() = %v, want nil", err) + } +} + +func TestServeListeningRefusesAStartWhileAnotherHoldsTheLock(t *testing.T) { + t.Parallel() + + path := socketAt(t) + path.holdLock(t) + path.boundSocket(t) + before := path.describe() + + listenRefused(t, path.listening(), "http server: another server is starting on "+string(path)) + + path.wantLockHeld(t, "the start that holds it") + if after := path.describe(); after != before { + t.Errorf("%s was %s and is now %s, want the socket of the start that holds the lock left in place", + path, before, after) + } +} + +func TestServeListeningHoldsTheLockUntilTheSocketIsGone(t *testing.T) { + t.Parallel() + + path := socketAt(t) + noForks(t) + run := serveListening(t, path.listening(), reportNames(), defaults) + + serving := path.tryLock() + err := run.end() + stopped := path.tryLock() + + if !errors.Is(serving, syscall.EWOULDBLOCK) { + t.Errorf("locking %s while the server served = %v, want it held", path.lockFile(), serving) + } + if err != nil || stopped != nil { + t.Errorf("ServeListening() = %v, locking once it returned = %v, want nil and the lock free", err, stopped) + } + if !path.gone() { + t.Errorf("%s is still there once ServeListening returned, want it removed", path) + } + if got := path.lockFile().describe(); got != `-rw-------, 0 bytes, linked to ""` { + t.Errorf("the lock file is %s, want it left in place, empty and of mode 0600", got) + } +} + +func TestServeListeningRefusesASocketAProgramWithoutTheLockServes(t *testing.T) { + t.Parallel() + + path := socketAt(t) + other, err := net.Listen("unix", string(path)) + if err != nil { + t.Fatalf("listening on %s: %v", path, err) + } + defer func() { _ = other.Close() }() + + listenRefused(t, path.listening(), "http server: a server already listens on "+string(path)) + + conn, err := net.Dial("unix", string(path)) + if err != nil { + t.Fatalf("the other program's socket at %s stopped answering: %v", path, err) + } + _ = conn.Close() +} + +// heldUntil returns a handler that tells arrived of its request, then answers once proceed closes. +func heldUntil(arrived chan<- struct{}, proceed <-chan struct{}) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + close(arrived) + <-proceed + _, _ = io.WriteString(w, "quarterly\n") + }) +} + +func TestServeListeningLetsANewStartServeWhileTheOldOneDrains(t *testing.T) { + t.Parallel() + + path := socketAt(t) + noForks(t) + arrived := make(chan struct{}) + proceed := make(chan struct{}) + release := sync.OnceFunc(func() { close(proceed) }) + defer release() + srv := gonsole.NewServer("", heldUntil(arrived, proceed), defaults) + unlistened := make(chan struct{}) + srv.RegisterOnShutdown(func() { close(unlistened) }) + old := attemptWith(t.Context(), srv, path.listening()) + if serving, err := old.settle(); !serving { + t.Fatalf("the old ServeListening() returned %v before listening", err) + } + answered := make(chan int, 1) + go func() { + code, _ := path.status() + answered <- code + }() + <-arrived + old.cancel() + <-unlistened + + fresh := serveListening(t, path.listening(), reportNames(), defaults) + body := path.get(t) + release() + + if code := <-answered; code != http.StatusOK { + t.Errorf("answer %d, want the request the old run drained answered 200", code) + } + if err := ended(t, old.done, defaults); err != nil { + t.Errorf("the old ServeListening() = %v, want nil", err) + } + if body != "quarterly\nyearly\n" { + t.Errorf("body = %q, want the new start answering the report names while the old run drained", body) + } + if err := fresh.end(); err != nil { + t.Errorf("the new ServeListening() = %v, want nil", err) + } +} + +func TestServeListeningStopsWhenTheSocketsFolderIsReadOnly(t *testing.T) { + t.Parallel() + + folder := shortFolder(t) + path := place(filepath.Join(folder, "myapp.sock")) + if err := os.WriteFile(string(path.lockFile()), nil, 0o600); err != nil { + t.Fatalf("writing %s: %v", path.lockFile(), err) + } + if err := os.Chmod(folder, 0o500); err != nil { + t.Fatalf("taking the write bit off %s: %v", folder, err) + } + t.Cleanup(func() { _ = os.Chmod(folder, 0o700) }) + + listenRefused(t, path.listening(), "http server: listen unix "+string(path)+": bind: permission denied") +} diff --git a/gonsole/socket_other.go b/gonsole/socket_other.go new file mode 100644 index 0000000..ecd5795 --- /dev/null +++ b/gonsole/socket_other.go @@ -0,0 +1,24 @@ +// SPDX-License-Identifier: Apache-2.0 + +//go:build !linux + +package gonsole + +import ( + "context" + "errors" + "net" +) + +// errNeedsLinux is the refusal of a socket listener off Linux. +var errNeedsLinux = errors.New("a socket listener needs Linux") + +// checkSocketPath refuses every socket path. +func checkSocketPath(string) error { + return errNeedsLinux +} + +// listenSocket refuses every socket listener. +func listenSocket(context.Context, string) (net.Listener, error) { + return nil, errNeedsLinux +} diff --git a/gonsole/socket_other_test.go b/gonsole/socket_other_test.go new file mode 100644 index 0000000..4b2f355 --- /dev/null +++ b/gonsole/socket_other_test.go @@ -0,0 +1,33 @@ +// SPDX-License-Identifier: Apache-2.0 + +//go:build !linux + +package gonsole_test + +import ( + "testing" + + "github.com/gopherium/framework/gonsole" +) + +func TestEnvRefusesASocketOffLinux(t *testing.T) { + t.Parallel() + + env := settings(map[string]string{ + "MYAPP_SOCKET": "/run/myapp/myapp.sock", "MYAPP_CLIENT_ADDR_HEADER": "X-Client-Addr", + }) + + got, err := env.Listening(local) + + if got != (gonsole.Listening{}) || errorText(err) != "MYAPP_SOCKET: a socket listener needs Linux" { + t.Errorf("Listening() = %+v, %q, want nothing and the refusal off Linux", got, errorText(err)) + } +} + +func TestServeListeningRefusesASocketOffLinux(t *testing.T) { + t.Parallel() + + at := gonsole.Listening{Network: "unix", Address: "/run/myapp/myapp.sock", ClientAddrHeader: "X-Client-Addr"} + + startRefused(t, at, defaults, "gonsole: Listening.Address: a socket listener needs Linux") +} diff --git a/gonsole/stall_internal_test.go b/gonsole/stall_internal_test.go new file mode 100644 index 0000000..6df7a4d --- /dev/null +++ b/gonsole/stall_internal_test.go @@ -0,0 +1,98 @@ +// SPDX-License-Identifier: Apache-2.0 + +package gonsole + +import ( + "context" + "io" + "log/slog" + "net" + "net/http" + "testing" + "testing/synctest" + "time" +) + +// stallBound is how much fake time a stalled body may hold a run before a test calls the run held. +const stallBound = time.Hour + +// readsTheBody answers each request once it has read the whole body. +func readsTheBody() http.Handler { + return http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { + _, _ = io.Copy(io.Discard, r.Body) + }) +} + +// stallABody sends over a new pipe a request that promises 1000 bytes of body and sends one, then reads every answer. +func stallABody(pipes *pipeListener) { + client := pipes.dial() + go func() { + _, _ = io.WriteString(client, "POST /reports HTTP/1.1\r\nHost: reports.example\r\nContent-Length: 1000\r\n\r\nq") + _, _ = io.Copy(io.Discard, client) + _ = client.Close() + }() +} + +// heldByAStall reports whether a stalled body holds a run of srv under timeouts for stallBound, idle after idleWindow. +func heldByAStall(t *testing.T, srv *http.Server, timeouts Timeouts) bool { + t.Helper() + parent, cancel := context.WithCancel(t.Context()) + defer cancel() + gone := make(chan struct{}) + srv.ConnState = func(_ net.Conn, state http.ConnState) { + if state == http.StateClosed { + close(gone) + } + } + pipes := newPipeListener() + run, listener, stop := windowed.idling(parent, srv, pipes, slog.New(slog.DiscardHandler)) + defer stop() + served := make(chan error, 1) + go func() { served <- serveOn(run, srv, listener, timeouts, nil, nil) }() + stallABody(pipes) + + time.Sleep(stallBound) + synctest.Wait() + held := len(served) == 0 + cancel() + <-served + <-gone + return held +} + +func TestEveryIdleExitServeListeningTakesEndsAStalledBody(t *testing.T) { + t.Parallel() + + bounded := Timeouts{ReadHeader: 10 * time.Second, Read: 30 * time.Second, Idle: 120 * time.Second, + Grace: time.Second, CancelGrace: time.Second, StopGrace: time.Second} + noRead := bounded + noRead.Read = 0 + noHeader := bounded + noHeader.ReadHeader = 0 + noIdle := bounded + noIdle.Idle = 0 + cases := []struct { + name string + timeouts Timeouts + }{ + {"the default timeouts", bounded}, + {"no read timeout", noRead}, + {"no header timeout", noHeader}, + {"no idle timeout", noIdle}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + srv := NewServer("", readsTheBody(), tc.timeouts) + if refuseIdle(windowed, srv) != nil { + return + } + if heldByAStall(t, srv, tc.timeouts) { + t.Errorf("ServeListening takes this idle exit, yet a stalled body held the run open for %v", stallBound) + } + }) + }) + } +} diff --git a/gonsole/visitor.go b/gonsole/visitor.go new file mode 100644 index 0000000..b0b7ec9 --- /dev/null +++ b/gonsole/visitor.go @@ -0,0 +1,64 @@ +// SPDX-License-Identifier: Apache-2.0 + +package gonsole + +import ( + "log/slog" + "net" + "net/http" + "net/netip" + "strings" +) + +// OverSocket reports whether r came over a Unix socket connection. +func OverSocket(r *http.Request) bool { + local, ok := r.Context().Value(http.LocalAddrContextKey).(net.Addr) + return ok && local.Network() == "unix" +} + +// visitors returns next with header dropped from every request and its address put in RemoteAddr on a socket. +func visitors(header string, next http.Handler, logger *slog.Logger) http.Handler { + if header == "" { + return next + } + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + values := r.Header.Values(header) + visitor := new(http.Request) + *visitor = *r + visitor.Header = r.Header.Clone() + visitor.Header.Del(header) + if !OverSocket(r) { + next.ServeHTTP(w, visitor) + return + } + address, ok := visitorAddr(values) + if !ok { + logger.Warn("refused a socket request without one visitor address", "header", header) + http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest) + return + } + visitor.RemoteAddr = address + next.ServeHTTP(w, visitor) + }) +} + +// visitorAddr returns the one address values hold as host and port, port 0 when it names none. +func visitorAddr(values []string) (string, bool) { + if len(values) != 1 || strings.Contains(values[0], ",") { + return "", false + } + visitor, err := parseVisitor(values[0]) + if err != nil { + return "", false + } + return netip.AddrPortFrom(visitor.Addr().Unmap().WithZone(""), visitor.Port()).String(), true +} + +// parseVisitor returns the address and port value names, port 0 for a bare address. +func parseVisitor(value string) (netip.AddrPort, error) { + if visitor, err := netip.ParseAddrPort(value); err == nil { + return visitor, nil + } + addr, err := netip.ParseAddr(value) + return netip.AddrPortFrom(addr, 0), err +}