Read a refused line's request before answering it, so a busy line is read as busy
The engine answered busy and closed before the client's request had landed; the client's write then broke and it never read the refusal waiting for it. Under the whole suite's load this failed the busy test (novox/hq issue 434).
This commit is contained in:
@@ -55,6 +55,11 @@ 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
|
||||
|
||||
// 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).
|
||||
@@ -247,31 +252,41 @@ 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 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.
|
||||
refuse := func(why string) {
|
||||
_ = conn.SetReadDeadline(time.Now().Add(min(refusedWithin, readWithin)))
|
||||
_, _ = cli.ReadRequest(conn)
|
||||
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)
|
||||
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)))
|
||||
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)
|
||||
|
||||
@@ -189,15 +189,20 @@ func TestOneAccountsLinesBeyondItsBoundAreAnsweredBusy(t *testing.T) {
|
||||
t.Cleanup(func() { close(release) })
|
||||
go func() { _, _ = cli.Ask(context.Background(), socket, []string{"push"}) }()
|
||||
<-started
|
||||
done := make(chan cli.Answer, 1)
|
||||
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 +212,50 @@ 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) })
|
||||
go func() { _, _ = cli.Ask(context.Background(), socket, []string{"push"}) }()
|
||||
<-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")
|
||||
}
|
||||
}
|
||||
|
||||
// **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) {
|
||||
|
||||
Reference in New Issue
Block a user