Merge pull request 'cli: a client reads the answer its failed write left waiting, so a refusal is heard (issue 444)' (#19) from fix/444-a-client-reads-the-answer-its-write-could-not-finish into main
This commit was merged in pull request #19.
This commit is contained in:
+15
-1
@@ -140,7 +140,21 @@ func AskWith(ctx context.Context, socket string, req Request) (Answer, error) {
|
||||
_ = conn.SetDeadline(deadline)
|
||||
}
|
||||
if err := WriteRequest(conn, req); err != nil {
|
||||
return Answer{}, fmt.Errorf("asking the node-engine: %w", err)
|
||||
// The engine may have answered before it read the request — a refusal (busy, an unknown account) written
|
||||
// and the connection closed — and the write then fails on the close. That answer is already waiting, and
|
||||
// it says why where the write error says only that the pipe broke (novox/hq issue 444, found under issue
|
||||
// 434). So read it, within the deadline the connection already has, and only when none came say the write.
|
||||
// Only a refusal is taken: the request was never wholly written, so nothing can have run, and any other
|
||||
// answer here is not one a correct engine gives — passing it on as success would hide the failure.
|
||||
a, rerr := ReadAnswer(conn)
|
||||
if rerr != nil {
|
||||
return Answer{}, fmt.Errorf("asking the node-engine: %w; and no answer came (%v)", err, rerr)
|
||||
}
|
||||
if a.Refused == "" {
|
||||
return Answer{}, fmt.Errorf("asking the node-engine: %w; and the answer that came is not a refusal, "+
|
||||
"though the request was never wholly written", err)
|
||||
}
|
||||
return a, nil
|
||||
}
|
||||
a, err := ReadAnswer(conn)
|
||||
if err != nil {
|
||||
|
||||
@@ -0,0 +1,123 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// anEngineThatClosesFirst listens like the node-engine before mesh-host#80: it writes its answer, if any, and
|
||||
// closes without reading the request (novox/hq issue 434). It answers one connection.
|
||||
func anEngineThatClosesFirst(t *testing.T, answer *Answer) string {
|
||||
t.Helper()
|
||||
socket := filepath.Join(t.TempDir(), "cli.sock")
|
||||
l, err := net.Listen("unix", socket)
|
||||
if err != nil {
|
||||
t.Skipf("no unix socket here: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { l.Close() })
|
||||
go func() {
|
||||
conn, err := l.Accept()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if answer != nil {
|
||||
_ = WriteAnswer(conn, *answer)
|
||||
}
|
||||
conn.Close()
|
||||
}()
|
||||
return socket
|
||||
}
|
||||
|
||||
// aLineTheSocketCannotHold is a request larger than a unix socket buffers, so its write cannot finish before
|
||||
// the engine reads it — and an engine that closes first makes it fail every time, not only when it loses a race.
|
||||
func aLineTheSocketCannotHold() []string { return []string{"status", strings.Repeat("x", 4<<20)} }
|
||||
|
||||
// An engine that refuses and closes before reading the request is heard: its refusal reaches the caller, not
|
||||
// the broken pipe its close left the client's write (novox/hq issue 444).
|
||||
func TestARefusalWrittenBeforeTheRequestIsReadReachesTheCaller(t *testing.T) {
|
||||
busy := Refusal("the node-engine is busy: too many lines at once")
|
||||
socket := anEngineThatClosesFirst(t, &busy)
|
||||
a, err := Ask(context.Background(), socket, aLineTheSocketCannotHold())
|
||||
if err != nil {
|
||||
t.Fatalf("the engine said %q and the caller heard: %v", busy.Refused, err)
|
||||
}
|
||||
if a.Refused != busy.Refused || a.Exit != 1 {
|
||||
t.Fatalf("the engine said %q and the caller heard %+v", busy.Refused, a)
|
||||
}
|
||||
}
|
||||
|
||||
// When the write fails and no answer is waiting, the error says both: what failed, and that no answer came.
|
||||
func TestAFailedWriteWithNoAnswerSaysBoth(t *testing.T) {
|
||||
socket := anEngineThatClosesFirst(t, nil)
|
||||
start := time.Now()
|
||||
_, err := Ask(context.Background(), socket, aLineTheSocketCannotHold())
|
||||
if err == nil {
|
||||
t.Fatal("a request the engine never read, and never answered, came back without an error")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "asking the node-engine") || !strings.Contains(err.Error(), "no answer came") {
|
||||
t.Fatalf("the error does not say what failed and that no answer came: %v", err)
|
||||
}
|
||||
if waited := time.Since(start); waited > 30*time.Second {
|
||||
t.Fatalf("waited %s on a connection the engine had closed", waited)
|
||||
}
|
||||
}
|
||||
|
||||
// After a failed write only a refusal is taken: nothing can have run, so an answer saying it ran and exited 0
|
||||
// is an error, never a success (novox/hq issue 444).
|
||||
func TestAnAnswerOtherThanARefusalAfterAFailedWriteIsAnError(t *testing.T) {
|
||||
ran := Answer{Exit: 0, Stdout: []byte("ok")}
|
||||
socket := anEngineThatClosesFirst(t, &ran)
|
||||
a, err := Ask(context.Background(), socket, aLineTheSocketCannotHold())
|
||||
if err == nil {
|
||||
t.Fatalf("a request never wholly written came back as success: %+v", a)
|
||||
}
|
||||
if !strings.Contains(err.Error(), "not a refusal") {
|
||||
t.Fatalf("the error does not say the answer was not a refusal: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Reading after a failed write never waits longer than the ask already allowed: a peer that never reads and
|
||||
// never answers is given up on at the context's deadline, not later (novox/hq issue 444).
|
||||
func TestAFailedWriteWaitsNoLongerThanTheDeadline(t *testing.T) {
|
||||
socket := filepath.Join(t.TempDir(), "cli.sock")
|
||||
l, err := net.Listen("unix", socket)
|
||||
if err != nil {
|
||||
t.Skipf("no unix socket here: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { l.Close() })
|
||||
held := make(chan net.Conn, 1)
|
||||
go func() {
|
||||
conn, err := l.Accept()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
held <- conn // neither read nor answered, nor closed, until the test ends
|
||||
}()
|
||||
t.Cleanup(func() {
|
||||
select {
|
||||
case c := <-held:
|
||||
c.Close()
|
||||
default:
|
||||
}
|
||||
})
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
|
||||
defer cancel()
|
||||
start := time.Now()
|
||||
_, err = Ask(ctx, socket, aLineTheSocketCannotHold())
|
||||
waited := time.Since(start)
|
||||
if err == nil {
|
||||
t.Fatal("a peer that never read nor answered gave no error")
|
||||
}
|
||||
if !errors.Is(err, os.ErrDeadlineExceeded) {
|
||||
t.Fatalf("the error is not the deadline: %v", err)
|
||||
}
|
||||
if waited > 3*time.Second {
|
||||
t.Fatalf("waited %s with a deadline of 1s", waited)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user