fix(serverless): keep a log follow alive until the stream opens

docs/runware_serverless_apps_logs.md

@@ -9,10 +9,12 @@ Show recent application logs, oldest first, and optionally follow new ones.
 The recent page is read from the runtime log query over --window (default 1h),
 and --limit and --cursor page through it. With --follow the command prints the
 recent page, then streams new entries until interrupted; the stream reconnects
-when the server ends it. The live stream has no window, so --window, --limit
-and --cursor apply to the recent page only, and --cursor cannot be combined
-with --follow. Entries written between the recent page and the start of the
-stream, or while the stream reconnects, can be missed or repeated.
+when the server ends it, and waits for the stream to open when a gateway
+answers first, which is what an application that has written nothing does. The
+live stream has no window, so --window, --limit and --cursor apply to the
+recent page only, and --cursor cannot be combined with --follow. Entries
+written between the recent page and the start of the stream, or while the
+stream reconnects, can be missed or repeated.
 
 In table format each entry is one line: time, level and message. In json or
 yaml format the recent page is printed as one document; with --follow every

internal/api/serverless/logs.go

@@ -53,6 +53,34 @@ func (e *TailStreamError) Error() string {
 	return "log stream failed: " + e.Detail
 }
 
+// TailUnavailableError reports that a gateway answered the tail request instead
+// of the log store, so the stream never opened. No entry was delivered, which
+// makes a reconnect free of repeated output. An app that has not written an
+// entry yet answers this way: the store holds the request without writing its
+// response headers, and the edge times it out before the first entry arrives.
+type TailUnavailableError struct {
+	StatusCode int
+	Err error
+}
+
+func (e *TailUnavailableError) Error() string {
+	return "log stream unavailable: " + e.Err.Error()
+}
+
+func (e *TailUnavailableError) Unwrap() error { return e.Err }
+
+// isGatewayStatus reports whether a proxy in front of the log store, rather
+// than the store itself, ended the request. These are transient, so a tail
+// reconnects on them instead of failing.
+func isGatewayStatus(statusCode int) bool {
+	switch statusCode {
+	case http.StatusBadGateway, http.StatusServiceUnavailable, http.StatusGatewayTimeout:
+ return true
+	default:
+ return false
+	}
+}
+
 // GetLogEntries returns one page of a named log query.
 func (c *Client) GetLogEntries(ctx context.Context, queryID string, params GetLogEntriesParams) (LogEntryPage, error) {
 	if c.apiKey == "" {
@@ -105,8 +133,9 @@ func (c *Client) GetLogEntries(ctx context.Context, queryID string, params GetLo
 
 // TailLogs follows a named log query for one app and hands every new entry to
 // emit. It returns ErrTailEnded when the server closes the stream cleanly, a
-// *TailStreamError when the server reports a failure, ctx.Err() when the
-// caller stops, and any error emit returns.
+// *TailStreamError when the server reports a failure, a *TailUnavailableError
+// when a gateway answers before the stream opens, ctx.Err() when the caller
+// stops, and any error emit returns.
 func (c *Client) TailLogs(ctx context.Context, queryID, appID string, emit func(LogEntry) error) error {
 	if c.apiKey == "" {
  return transport.ErrNoAPIKey
@@ -127,7 +156,14 @@ func (c *Client) TailLogs(ctx context.Context, queryID, appID string, emit func(
 	if resp.StatusCode != http.StatusOK {
  body, _ := io.ReadAll(io.LimitReader(resp.Body, maxSSEFrameBytes))
  c.logResponse(ctx, resp, body)
- return problemFromBody(body, resp.StatusCode)
+ problem := problemFromBody(body, resp.StatusCode)
+ if isGatewayStatus(resp.StatusCode) {
+ return &TailUnavailableError{
+ StatusCode: resp.StatusCode,
+ Err: problem,
+ }
+ }
+ return problem
 	}
 	c.logResponse(ctx, resp, nil)
 	if mediaType, _, _ := mime.ParseMediaType(resp.Header.Get("Content-Type")); mediaType != "text/event-stream" {

internal/api/serverless/logs_test.go

@@ -247,6 +247,53 @@ func TestTailLogs_NonStreamStatusIsAProblem(t *testing.T) {
 	}
 }
 
+func TestTailLogs_GatewayStatusIsRetryable(t *testing.T) {
+	for _, status := range []int{http.StatusBadGateway, http.StatusServiceUnavailable, http.StatusGatewayTimeout} {
+ t.Run(http.StatusText(status), func(t *testing.T) {
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+ // What the edge answers with: plain text, not a problem document.
+ w.Header().Set("Content-Type", "text/plain; charset=UTF-8")
+ w.WriteHeader(status)
+ _, _ = fmt.Fprintf(w, "error code: %d", status)
+ }))
+ defer srv.Close()
+
+ c := newClient("test-key", srv.URL, slog.Default(), srv.Client())
+ err := c.TailLogs(context.Background(), LogQueryRuntimeTail, testAppID, func(LogEntry) error { return nil })
+ unavailable, ok := errors.AsType[*TailUnavailableError](err)
+ if !ok || unavailable.StatusCode != status {
+ t.Fatalf("err = %#v", err)
+ }
+ if !strings.HasPrefix(unavailable.Error(), "log stream unavailable: ") {
+ t.Fatalf("message = %q", unavailable.Error())
+ }
+ re, ok := errors.AsType[*transport.RunwareError](err)
+ if !ok || re.StatusCode != status {
+ t.Fatalf("unwrapped err = %#v", err)
+ }
+ })
+	}
+}
+
+func TestTailLogs_NonGatewayServerErrorStaysFatal(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+ w.Header().Set("Content-Type", testLogsProblemJSON)
+ w.WriteHeader(http.StatusInternalServerError)
+ _, _ = w.Write([]byte(`{"title":"Internal Server Error","status":500,"detail":"the log store failed"}`))
+	}))
+	defer srv.Close()
+
+	c := newClient("test-key", srv.URL, slog.Default(), srv.Client())
+	err := c.TailLogs(context.Background(), LogQueryRuntimeTail, testAppID, func(LogEntry) error { return nil })
+	if _, ok := errors.AsType[*TailUnavailableError](err); ok {
+ t.Fatalf("500 must not be retryable, err = %#v", err)
+	}
+	re, ok := errors.AsType[*transport.RunwareError](err)
+	if !ok || re.StatusCode != http.StatusInternalServerError {
+ t.Fatalf("err = %#v", err)
+	}
+}
+
 func TestTailLogs_RejectsNonEventStreamBody(t *testing.T) {
 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
  w.Header().Set("Content-Type", "application/json")

internal/cmd/serverless/apps_logs.go

@@ -20,8 +20,9 @@ import (
 )
 
 // tailReconnectDelay separates two connection attempts after the server
-// reports a stream failure, or after a clean end of a stream that lived
-// shorter than this, so a server that closes at once is not hammered.
+// reports a stream failure, after a gateway answers before the stream opens,
+// or after a clean end of a stream that lived shorter than this, so a server
+// that closes at once is not hammered.
 const tailReconnectDelay = 2 * time.Second
 
 // logWindows lists the accepted --window values, in the order they are documented.
@@ -50,10 +51,12 @@ func newAppsLogsCmd(logger *log.Logger) *cobra.Command {
 The recent page is read from the runtime log query over --window (default 1h),
 and --limit and --cursor page through it. With --follow the command prints the
 recent page, then streams new entries until interrupted; the stream reconnects
-when the server ends it. The live stream has no window, so --window, --limit
-and --cursor apply to the recent page only, and --cursor cannot be combined
-with --follow. Entries written between the recent page and the start of the
-stream, or while the stream reconnects, can be missed or repeated.
+when the server ends it, and waits for the stream to open when a gateway
+answers first, which is what an application that has written nothing does. The
+live stream has no window, so --window, --limit and --cursor apply to the
+recent page only, and --cursor cannot be combined with --follow. Entries
+written between the recent page and the start of the stream, or while the
+stream reconnects, can be missed or repeated.
 
 In table format each entry is one line: time, level and message. In json or
 yaml format the recent page is printed as one document; with --follow every
@@ -187,21 +190,34 @@ func logEmitter(format output.Format, out io.Writer) func(serverlessapi.LogEntry
 }
 
 // followLogs keeps a live stream open until ctx is cancelled. A clean end of a
-// long-lived stream reconnects at once; a reported stream failure, or a clean
-// end of a short-lived stream, reconnects after tailReconnectDelay. Any other
-// error is returned. Cancellation is a normal exit.
+// long-lived stream reconnects at once; a reported stream failure, a gateway
+// answering before the stream opens, or a clean end of a short-lived stream,
+// reconnects after tailReconnectDelay. Any other error is returned.
+// Cancellation is a normal exit.
 func followLogs(ctx context.Context, tail logTailer, emit func(serverlessapi.LogEntry) error, errOut io.Writer) error {
+	reportedUnavailable := false
 	for {
  started := time.Now()
  err := tail(ctx, emit)
  if ctx.Err() != nil {
  return nil //nolint:nilerr // Cancellation is the normal way a follow ends.
  }
  _, streamFailed := errors.AsType[*serverlessapi.TailStreamError](err)
+ unavailable, gatewayAnswered := errors.AsType[*serverlessapi.TailUnavailableError](err)
  switch {
  case streamFailed:
+ reportedUnavailable = false
  _, _ = fmt.Fprintf(errOut, "%v; reconnecting\n", err)
+ case gatewayAnswered:
+ // Every attempt on an app that has written nothing answers this
+ // way, so the notice would repeat for as long as the app stays
+ // quiet. Say it once, then wait in silence for the stream to open.
+ if !reportedUnavailable {
+ reportedUnavailable = true
+ _, _ = fmt.Fprintf(errOut, "%v; waiting for the stream to open\n", unavailable)
+ }
  case errors.Is(err, serverlessapi.ErrTailEnded):
+ reportedUnavailable = false
  if time.Since(started) >= tailReconnectDelay {
  continue
  }

internal/cmd/serverless/apps_logs_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"errors"
+	"net/http"
 	"strings"
 	"testing"
 	"testing/synctest"
@@ -231,6 +232,88 @@ func TestFollowLogs_WaitsBeforeReconnectingAfterStreamFailure(t *testing.T) {
 	})
 }
 
+func TestFollowLogs_WaitsForTheStreamToOpenWhenAGatewayAnswers(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ var attempts []time.Time
+ var emitted []string
+ tail := func(_ context.Context, emit func(serverlessapi.LogEntry) error) error {
+ attempts = append(attempts, time.Now())
+ // Three gateway answers, as a quiet application produces, then the
+ // stream opens and carries one entry.
+ if len(attempts) <= 3 {
+ return &serverlessapi.TailUnavailableError{
+ StatusCode: http.StatusGatewayTimeout,
+ Err: errors.New("the origin did not respond in time"),
+ }
+ }
+ if err := emit(serverlessapi.LogEntry{Body: testLogBodyReady}); err != nil {
+ return err
+ }
+ cancel()
+ return context.Canceled
+ }
+ var errOut bytes.Buffer
+ emit := func(entry serverlessapi.LogEntry) error {
+ emitted = append(emitted, entry.Body)
+ return nil
+ }
+ if err := followLogs(ctx, tail, emit, &errOut); err != nil {
+ t.Fatalf("followLogs: %v", err)
+ }
+ if len(attempts) != 4 {
+ t.Fatalf("attempts = %v", attempts)
+ }
+ for i := 1; i < len(attempts); i++ {
+ if attempts[i].Sub(attempts[i-1]) != tailReconnectDelay {
+ t.Fatalf("attempt %d waited %v", i, attempts[i].Sub(attempts[i-1]))
+ }
+ }
+ if len(emitted) != 1 || emitted[0] != testLogBodyReady {
+ t.Fatalf("emitted = %v", emitted)
+ }
+ // The notice names the condition once, however long the wait lasts.
+ if got := strings.Count(errOut.String(), "waiting for the stream to open"); got != 1 {
+ t.Fatalf("notice count = %d, stderr = %q", got, errOut.String())
+ }
+ if !strings.Contains(errOut.String(), "log stream unavailable: the origin did not respond in time") {
+ t.Fatalf("stderr = %q", errOut.String())
+ }
+	})
+}
+
+func TestFollowLogs_NoticesTheGatewayAgainAfterAStreamOpened(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ attempts := 0
+ unavailable := func() error {
+ return &serverlessapi.TailUnavailableError{
+ StatusCode: http.StatusBadGateway,
+ Err: errors.New("bad gateway"),
+ }
+ }
+ tail := func(context.Context, func(serverlessapi.LogEntry) error) error {
+ attempts++
+ switch attempts {
+ case 1, 3:
+ return unavailable()
+ case 2:
+ return serverlessapi.ErrTailEnded
+ default:
+ cancel()
+ return context.Canceled
+ }
+ }
+ var errOut bytes.Buffer
+ if err := followLogs(ctx, tail, func(serverlessapi.LogEntry) error { return nil }, &errOut); err != nil {
+ t.Fatalf("followLogs: %v", err)
+ }
+ if got := strings.Count(errOut.String(), "waiting for the stream to open"); got != 2 {
+ t.Fatalf("notice count = %d, stderr = %q", got, errOut.String())
+ }
+	})
+}
+
 func TestFollowLogs_ReturnsOtherErrors(t *testing.T) {
 	boom := errors.New("boom")
 	tail := func(context.Context, func(serverlessapi.LogEntry) error) error { return boom }