diff --git a/internal/meshcli/meshcli.go b/internal/meshcli/meshcli.go index f1cd5f5..31a4b50 100644 --- a/internal/meshcli/meshcli.go +++ b/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())) diff --git a/internal/meshcli/meshcli_test.go b/internal/meshcli/meshcli_test.go index 73c252a..4b83f51 100644 --- a/internal/meshcli/meshcli_test.go +++ b/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) {