Bound what a refusal costs the engine, and let the busy tests fail instead of hang

A refusal waiting for its request parsed it into a 64 KiB reader for up to 5 s,
on a socket every account may open. It now drains at most 4 KiB through a small
buffer for at most 1 s and never past the accept-time deadline, only 32 wait at
once, and an unknown account is refused before it takes a slot. The busy tests
wait for the first line with a timeout and say what came of it (novox/hq issue 434).

internal/meshcli/meshcli.go

@@ -11,9 +11,11 @@
 package meshcli
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"net"
 	"os"
 	"os/user"
@@ -55,10 +57,32 @@ type Forward func(ctx context.Context, asked Asked) cli.Answer
 // holds nothing for long. A variable so a test need not wait.
 var readWithin = 30 * time.Second
 
-// refusedWithin bounds how long the engine reads the request of a line it refuses before it is asked: long enough
-// for a client that has connected to finish writing, short enough that a refused connection holds nothing for long.
-// A variable so a test can say.
-var refusedWithin = 5 * time.Second
+// refusedWithin bounds how long the engine waits for the request of a line it refuses before it is asked: a client
+// that has connected writes its request within microseconds, so a refused connection holds nothing for long. Never
+// longer than readWithin. A variable so a test can say.
+var refusedWithin = 1 * time.Second
+
+// refusalsWaiting bounds how many refusals wait for their request at the same time, on the whole engine; drainCap
+// bounds how much of a refused request is read, and it is read into a small buffer and dropped. The socket is open
+// to every account, so a refusal must cost the engine little however many connections ask (review of ADR 0272).
+// Variables so a test can say.
+var (
+	refusalsWaiting = make(chan struct{}, 32)
+	drainCap        = 4 << 10
+)
+
+// drain reads and drops what a refused client wrote, up to the end of its request line, the end of the stream, the
+// connection's deadline or drainCap bytes, whichever comes first. Nothing is kept.
+func drain(r io.Reader) {
+	var buf [512]byte
+	for read := 0; read < drainCap; {
+		n, err := r.Read(buf[:min(len(buf), drainCap-read)])
+		read += n
+		if bytes.IndexByte(buf[:n], '\n') >= 0 || err != nil {
+			return
+		}
+	}
+}
 
 // atOnce bounds how many lines are asked at the same time, and perAccount how many of them one account's: a person
 // types one at a time, so more is a program on the machine. One over either bound is answered busy at once — the
@@ -234,17 +258,18 @@ func Serve(ctx context.Context, socket string, forward Forward, say func(string)
 		}
 		// Bounded from the first byte: nothing below may wait on a connection for longer than this before the
 		// line is read, whatever the peer does (review of ADR 0272).
-		_ = conn.SetDeadline(time.Now().Add(readWithin))
+		readBy := time.Now().Add(readWithin)
+		_ = conn.SetDeadline(readBy)
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			answer(ctx, conn, held, forward, say)
+			answer(ctx, conn, readBy, held, forward, say)
 		}()
 	}
 }
 
-// answer serves one connection.
-func answer(ctx context.Context, conn *net.UnixConn, held *slots, forward Forward, say func(string)) {
+// answer serves one connection; readBy is the deadline set when it was accepted, for reading its line.
+func answer(ctx context.Context, conn *net.UnixConn, readBy time.Time, held *slots, forward Forward, say func(string)) {
 	defer conn.Close()
 	reply := func(a cli.Answer) {
 		_ = conn.SetWriteDeadline(time.Now().Add(readWithin))
@@ -255,11 +280,22 @@ func answer(ctx context.Context, conn *net.UnixConn, held *slots, forward Forwar
 	// A refusal is written only after the request it answers is read (novox/hq issue 434). The client writes its
 	// request first and reads the answer after; an engine that answers and closes before that write has landed
 	// makes the write fail with a broken pipe, and the client never reads the refusal waiting for it — under load a
-	// line beyond its account's bound read as an error instead of busy. The read is bounded by refusedWithin, and
-	// what it says is never asked; nor does it outlast readWithin, which bounds every wait before the line is read.
+	// line beyond its account's bound read as an error instead of busy. The wait ends at refusedWithin, or earlier
+	// at the deadline set when the connection was accepted; what was written is drained and never asked. Only
+	// refusalsWaiting refusals wait at once: over that the refusal is written at once, as before, and only then can a
+	// client still find its write broken.
 	refuse := func(why string) {
-		_ = conn.SetReadDeadline(time.Now().Add(min(refusedWithin, readWithin)))
-		_, _ = cli.ReadRequest(conn)
+		select {
+		case refusalsWaiting <- struct{}{}:
+			by := time.Now().Add(refusedWithin)
+			if readBy.Before(by) {
+				by = readBy
+			}
+			_ = conn.SetReadDeadline(by)
+			drain(conn)
+			<-refusalsWaiting
+		default:
+		}
 		reply(cli.Refusal(why))
 	}
 	uid, pid, err := peerOf(conn)
@@ -278,17 +314,18 @@ func answer(ctx context.Context, conn *net.UnixConn, held *slots, forward Forwar
 	// And the login session, now: judged as the peer was when it connected, before it can exec or exit while its
 	// line is read (review of ADR 0272).
 	session := sessionOf(pid, uid)
+	// Looked up before a slot is taken, so a refused account never holds one while it is refused.
+	account, err := lookupAccount(uid)
+	if err != nil || account == "" {
+		refuse(fmt.Sprintf("user id %d is not in this machine's user database, so nothing was asked", uid))
+		return
+	}
 	if !held.take(uid) {
 		refuse(fmt.Sprintf("busy: user id %d has %d lines being asked already, or the machine %d; ask "+
 			"again when one has ended. Nothing was asked", uid, perAccount, atOnce))
 		return
 	}
 	defer held.give(uid)
-	account, err := lookupAccount(uid)
-	if err != nil || account == "" {
-		refuse(fmt.Sprintf("user id %d is not in this machine's user database, so nothing was asked", uid))
-		return
-	}
 	req, err := cli.ReadRequest(conn)
 	if err != nil {
 		reply(cli.Refusal("the request could not be read, so nothing was asked: " + err.Error()))

internal/meshcli/meshcli_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"io"
 	"net"
 	"os"
 	"os/user"
@@ -172,6 +173,25 @@ func keysOf(t *testing.T, body []byte) string {
 	return strings.Join(keys, " ")
 }
 
+// holdALine asks one line and returns once the engine is asking it, so it holds its account's slot. A first line
+// that ends or never arrives fails the test at once, with what came of it, instead of hanging the suite (novox/hq
+// issue 434).
+func holdALine(t *testing.T, socket string, started <-chan struct{}) {
+	t.Helper()
+	first := make(chan error, 1)
+	go func() {
+		_, err := cli.Ask(context.Background(), socket, []string{"push"})
+		first <- err
+	}()
+	select {
+	case <-started:
+	case err := <-first:
+		t.Fatalf("the first line ended before it was asked: %v", err)
+	case <-time.After(5 * time.Second):
+		t.Fatal("the first line was never asked")
+	}
+}
+
 // **One account cannot hold the socket** (review of ADR 0272): its lines beyond its own bound are answered busy at
 // once, nothing is asked, and the engine goes on accepting.
 func TestOneAccountsLinesBeyondItsBoundAreAnsweredBusy(t *testing.T) {
@@ -187,8 +207,7 @@ func TestOneAccountsLinesBeyondItsBoundAreAnsweredBusy(t *testing.T) {
 	})
 	// Released before the engine is stopped: cleanups run last first, and the engine waits for its lines.
 	t.Cleanup(func() { close(release) })
-	go func() { _, _ = cli.Ask(context.Background(), socket, []string{"push"}) }()
-	<-started
+	holdALine(t, socket, started)
 	type asked struct {
 		a   cli.Answer
 		err error
@@ -229,8 +248,7 @@ func TestARefusalWaitsForTheRequestItAnswers(t *testing.T) {
 		return cli.Answer{}
 	})
 	t.Cleanup(func() { close(release) })
-	go func() { _, _ = cli.Ask(context.Background(), socket, []string{"push"}) }()
-	<-started
+	holdALine(t, socket, started)
 	conn, err := net.Dial("unix", socket)
 	if err != nil {
 		t.Fatal(err)
@@ -256,6 +274,62 @@ func TestARefusalWaitsForTheRequestItAnswers(t *testing.T) {
 	}
 }
 
+// **Refusals that wait are bounded** (novox/hq issue 434, review of ADR 0272): with refusalsWaiting already full, a
+// further refused line is answered at once, without waiting for its request.
+func TestARefusalOverTheWaitingBoundIsAnsweredAtOnce(t *testing.T) {
+	was, wasWithin, wasWaiting := perAccount, refusedWithin, refusalsWaiting
+	perAccount, refusedWithin = 1, time.Minute
+	refusalsWaiting = make(chan struct{}, 1)
+	refusalsWaiting <- struct{}{}
+	t.Cleanup(func() { perAccount, refusedWithin, refusalsWaiting = was, wasWithin, wasWaiting })
+	release := make(chan struct{})
+	started := make(chan struct{}, 2)
+	socket := serving(t, func(context.Context, Asked) cli.Answer {
+		started <- struct{}{}
+		<-release
+		return cli.Answer{}
+	})
+	t.Cleanup(func() { close(release) })
+	holdALine(t, socket, started)
+	conn, err := net.Dial("unix", socket)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+	// Nothing is written: a refusal that waited would wait refusedWithin, a minute, for it.
+	_ = conn.SetDeadline(time.Now().Add(5 * time.Second))
+	a, err := cli.ReadAnswer(conn)
+	if err != nil || !strings.Contains(a.Refused, "busy") {
+		t.Fatalf("a refusal over the waiting bound reads %+v (%v)", a, err)
+	}
+}
+
+// countingReader is an endless request with no line end, counting what was read of it.
+type countingReader struct{ read int }
+
+func (c *countingReader) Read(p []byte) (int, error) {
+	for i := range p {
+		p[i] = 'x'
+	}
+	c.read += len(p)
+	return len(p), nil
+}
+
+// **A refused request is drained, not kept** (novox/hq issue 434): what a refused client writes is read up to
+// drainCap bytes and no further, however much it sends.
+func TestARefusedRequestIsDrainedNoFurtherThanItsCap(t *testing.T) {
+	r := &countingReader{}
+	drain(r)
+	if r.read != drainCap {
+		t.Fatalf("drained %d bytes of an endless request; the cap is %d", r.read, drainCap)
+	}
+	after := &countingReader{}
+	drain(io.MultiReader(strings.NewReader("{\"line\":[\"status\"]}\n"), after))
+	if after.read != 0 {
+		t.Fatalf("draining read %d bytes past the end of the request line", after.read)
+	}
+}
+
 // **A process in another namespace is not this machine's account** (review of ADR 0272): a container running with
 // the operator's user id shares the id and not the machine; refused, and nothing is asked.
 func TestAPeerInAnotherNamespaceIsRefused(t *testing.T) {