novox/mesh-host
6
0
Fork 0
Code Issues Pull Requests 2 Releases Wiki Activity

Merge pull request 'A refused line's request is read before it is answered, so a busy line reads busy (issue 434)' (#80) from fix/434-a-busy-line-is-answered-busy-whatever-the-load into main

Browse Source
This commit was merged in pull request #80.
This commit is contained in:
mesh-admin
2026-10-11 00:55:47 +00:00
parent bd5cc69804 f9805b2113
commit 85c51ebb59
2 changed files with 197 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/meshcli/meshcli.go
+66 -14
View File
@@ -11,9 +11,11 @@
package meshcli
import (
"bytes"
"context"
"errors"
"fmt"
"io"
"net"
"os"
"os/user"
@@ -55,6 +57,33 @@ 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 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
// engine never stops accepting for it, so one account cannot hold the socket from every other (review of ADR 0272).
@@ -229,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))
@@ -247,33 +277,55 @@ func answer(ctx context.Context, conn *net.UnixConn, held *slots, forward Forwar
say(fmt.Sprintf("mesh-cli: the answer could not be written back: %v", err))
}
}
// 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 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) {
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)
if err != nil {
// Never sent as anyone: an account the kernel did not give is no account at all.
reply(cli.Refusal("the node-engine could not read which account is asking, so nothing was asked: " + err.Error()))
refuse("the node-engine could not read which account is asking, so nothing was asked: " + err.Error())
return
}
// The user id is this machine's account only in this machine's namespaces: a container running with the
// operator's id shares the number and nothing else (review of ADR 0272).
if err := sameNamespaces(pid); err != nil {
reply(cli.Refusal("the asking process does not run in this machine's own namespaces, so it is not this " +
"machine's account and nothing was asked: " + err.Error()))
refuse("the asking process does not run in this machine's own namespaces, so it is not this " +
"machine's account and nothing was asked: " + err.Error())
return
}
// 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) {
reply(cli.Refusal(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)))
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 == "" {
reply(cli.Refusal(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
+131 -8
View File
@@ -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,17 +207,21 @@ 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
done := make(chan cli.Answer, 1)
holdALine(t, socket, started)
type asked struct {
a cli.Answer
err error
}
done := make(chan asked, 1)
go func() {
a, _ := cli.Ask(context.Background(), socket, []string{"status"})
done <- a
a, err := cli.Ask(context.Background(), socket, []string{"status"})
done <- asked{a, err}
}()
select {
case a := <-done:
if !strings.Contains(a.Refused, "busy") {
t.Fatalf("a second line from the same account reads %+v", a)
case got := <-done:
// An error is a busy line the client never read (novox/hq issue 434), not an answer.
if got.err != nil || !strings.Contains(got.a.Refused, "busy") {
t.Fatalf("a second line from the same account reads %+v (%v)", got.a, got.err)
}
case <-time.After(5 * time.Second):
t.Fatal("a second line from the same account waited instead of being answered busy")
@@ -207,6 +231,105 @@ func TestOneAccountsLinesBeyondItsBoundAreAnsweredBusy(t *testing.T) {
}
}
// **A refusal waits for the request it answers** (novox/hq issue 434): the client writes its line first and reads
// the answer after. An engine that answered busy and closed before that write landed broke the client's write, and
// the refusal was never read — under the whole suite's load the busy line read as an error. Here the client
// connects and holds its line back: no answer may come before the line, and the line, once written, is answered
// busy. Ordered by the client's own write, so it holds however slow the machine is.
func TestARefusalWaitsForTheRequestItAnswers(t *testing.T) {
was, wasWithin := perAccount, refusedWithin
perAccount, refusedWithin = 1, time.Minute
t.Cleanup(func() { perAccount, refusedWithin = was, wasWithin })
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 yet: an engine that answers now answers before the request. The wait is far inside
// refusedWithin, so an engine that waits is still waiting when it ends.
_ = conn.SetReadDeadline(time.Now().Add(200 * time.Millisecond))
var b [1]byte
if n, err := conn.Read(b[:]); n > 0 || !errors.Is(err, os.ErrDeadlineExceeded) {
t.Fatalf("the engine answered before the request was written (%d bytes, %v)", n, err)
}
_ = conn.SetDeadline(time.Now().Add(5 * time.Second))
if err := cli.WriteRequest(conn, cli.Request{Line: []string{"status"}}); err != nil {
t.Fatalf("the request could not be written: %v", err)
}
a, err := cli.ReadAnswer(conn)
if err != nil || !strings.Contains(a.Refused, "busy") {
t.Fatalf("a line written after the engine chose busy reads %+v (%v)", a, err)
}
if len(started) != 0 {
t.Fatal("the busy line was asked")
}
}
// **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) {