fix(nomad): send the error of a broken file stream before its lines close

internal/nomad/agentlogs.go

@@ -51,17 +51,7 @@ func (c *Client) followAgent(target AgentTarget, level string, lines chan<- stri
 
 	frames, failed := c.api.Agent().Monitor(cancel, c.agentQuery(target, map[string]string{"log_level": level}))
 
-	if frames != nil {
-		for chunk := range streamLines(frames, notCut, nil, nil, cancel) {
-			if !sendText(lines, chunk, cancel) {
-				return
-			}
-		}
-	}
-
-	// The error goes out after the lines and before they close, so a reader
-	// that sees the end finds it.
-	passFailures(failed, errs, cancel)
+	followFrames(frames, failed, notCut, nil, lines, errs, cancel)
 }
 
 // agentPast reads the log file of the agent and keeps its last MiB at the

internal/nomad/files.go

@@ -68,7 +68,8 @@ var ErrNotText = errors.New("not text")
 
 // File follows a file of an allocation as it grows: from its start, or when
 // it is big, from its last MiB. The caller closes the stream when it stops
-// reading, otherwise the request stays open. Do not call it for a pipe: the
+// reading, otherwise the request stays open. A failure arrives on its Err after
+// the lines before it and before Lines closes. Do not call it for a pipe: the
 // client reads its first bytes to detect the content type, and waits.
 func (c *Client) File(ctx context.Context, namespace, allocID, name string) (*LogStream, error) {
 	// The stream is requested from the node that runs the allocation when it
@@ -98,22 +99,35 @@ func (c *Client) File(ctx context.Context, namespace, allocID, name string) (*Lo
 		atEnd = func(frame *api.StreamFrame) bool { return frame.Offset >= info.Size }
 	}
 
-	lines := streamLines(frames, func(*api.StreamFrame) bool { return from > 0 }, atEnd, nil, cancel)
+	lines := make(chan string)
+	out := make(chan error, 1)
 
-	return &LogStream{Lines: lines, Err: failures(errs, cancel), Size: info.Size, From: from, cancel: cancel}, nil
-}
+	go func() {
+		defer close(lines)
 
-// failures passes on the errors of a stream. Its end is not a failure: the
-// stream of a file reports it as an error, the stream of a log does not.
-func failures(errs <-chan error, cancel <-chan struct{}) <-chan error {
-	out := make(chan error, 1)
+		followFrames(frames, errs, func(*api.StreamFrame) bool { return from > 0 }, atEnd, lines, out, cancel)
+	}()
 
-	go passFailures(errs, out, cancel)
+	return &LogStream{Lines: lines, Err: out, Size: info.Size, From: from, cancel: cancel}, nil
+}
+
+// followFrames sends the lines of the frames, then the first failure. The
+// failure goes out after the lines and before they close (the caller closes
+// them when this returns), so a reader that sees the end finds it.
+func followFrames(frames <-chan *api.StreamFrame, failed <-chan error, startsCut, atEnd func(*api.StreamFrame) bool, lines chan<- string, errs chan<- error, cancel <-chan struct{}) {
+	if frames != nil {
+		for chunk := range streamLines(frames, startsCut, atEnd, nil, cancel) {
+			if !sendText(lines, chunk, cancel) {
+				return
+			}
+		}
+	}
 
-	return out
+	passFailures(failed, errs, cancel)
 }
 
-// passFailures sends the first error of errs to out, unless it is an end.
+// passFailures sends the first error of errs to out, unless it is an end. The
+// stream of a file reports its end as an error, the stream of a log does not.
 func passFailures(errs <-chan error, out chan<- error, cancel <-chan struct{}) {
 	select {
 	case err := <-errs:

internal/nomad/files_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"slices"
@@ -259,3 +260,55 @@ func TestFile_WhatIsNotText(t *testing.T) {
 		r.NotEqual("/v1/client/fs/stream/af1f37df", req.path)
 	}
 }
+
+func TestFile_ABreakIsItsErrorAfterItsLines(t *testing.T) {
+	r := require.New(t)
+
+	// The body ends inside a frame: the decoder fails after the last line.
+	client, _ := clusterServer(t, map[string]string{
+		"/v1/allocation/af1f37df":       allocInfo,
+		"/v1/client/fs/stat/af1f37df":   `{"Name": "app.env", "Size": 4, "FileMode": "-rw-r--r--", "ContentType": "text/plain"}`,
+		"/v1/client/fs/stream/af1f37df": streamed(t, 4, "a=1\n") + `{"Data":"`,
+	})
+
+	stream, err := client.File(context.Background(), "production", "af1f37df", "/t/local/app.env")
+	r.NoError(err)
+	t.Cleanup(stream.Close)
+
+	select {
+	case err := <-stream.Err:
+		t.Fatalf("the error came before the last line: %v", err)
+	case <-time.After(200 * time.Millisecond):
+	}
+
+	r.Equal("a=1\n", readAll(t, stream))
+
+	select {
+	case err := <-stream.Err:
+		r.ErrorIs(err, io.ErrUnexpectedEOF)
+	default:
+		t.Fatal("the lines ended before the error")
+	}
+}
+
+func TestFile_ARefusedStreamIsItsError(t *testing.T) {
+	r := require.New(t)
+
+	client, _ := clusterServer(t, map[string]string{
+		"/v1/allocation/af1f37df":     allocInfo,
+		"/v1/client/fs/stat/af1f37df": `{"Name": "app.env", "Size": 4, "FileMode": "-rw-r--r--", "ContentType": "text/plain"}`,
+	})
+
+	stream, err := client.File(context.Background(), "production", "af1f37df", "/t/local/app.env")
+	r.NoError(err)
+	t.Cleanup(stream.Close)
+
+	select {
+	case err := <-stream.Err:
+		r.ErrorContains(err, "404")
+	case <-time.After(2 * time.Second):
+		t.Fatal("no error was sent")
+	}
+
+	r.Empty(readAll(t, stream))
+}

internal/nomad/logs_stream_test.go

@@ -1,6 +1,7 @@
 package nomad
 
 import (
+	"io"
 	"strings"
 	"testing"
 	"time"
@@ -291,3 +292,49 @@ func TestStreamLines_HoldALogLineUntilItsBreakWhileFramesStayOpen(t *testing.T)
 
 	require.Equal(t, []string{"waiting done\n"}, collect(t, lines, 1))
 }
+
+func TestFollowFrames_TheLinesWaitForAnErrorThatComesAfterTheFrames(t *testing.T) {
+	r := require.New(t)
+
+	frames := make(chan *api.StreamFrame, 1)
+	failed := make(chan error, 1)
+	lines := make(chan string)
+	errs := make(chan error, 1)
+	cancel := make(chan struct{})
+
+	t.Cleanup(func() { close(cancel) })
+
+	go func() {
+		defer close(lines)
+
+		followFrames(frames, failed, neverCut, nil, lines, errs, cancel)
+	}()
+
+	frames <- &api.StreamFrame{Data: []byte("a=1\n")}
+	close(frames)
+
+	select {
+	case line := <-lines:
+		r.Equal("a=1\n", line)
+	case <-time.After(2 * time.Second):
+		t.Fatal("no line was sent")
+	}
+
+	// The frames are closed and the error is not here yet: the lines stay open.
+	select {
+	case line, ok := <-lines:
+		t.Fatalf("the lines sent %q (open %v) before the error", line, ok)
+	case <-time.After(100 * time.Millisecond):
+	}
+
+	failed <- io.ErrUnexpectedEOF
+
+	r.True(ended(lines))
+
+	select {
+	case err := <-errs:
+		r.ErrorIs(err, io.ErrUnexpectedEOF)
+	default:
+		t.Fatal("the lines ended before the error")
+	}
+}