fix(appproto): free the passthrough slot before an answer's last frame

.changeset/api-passthrough-free-before-end.md

@@ -0,0 +1,5 @@
+---
+"ftw": patch
+---
+
+The box now takes the app's next API request as soon as the previous answer has ended. Before, it freed its one request slot only after it had sent the answer's last frame. An app that asked again at once could be told the box was busy and then wait out a backoff before the view filled in.

go/internal/api/api_passthrough_test.go

@@ -571,13 +571,18 @@ func TestReadingTheConfigIsAllowedForAnyRole(t *testing.T) {
 				t.Fatalf("headers = %v, want the content type", head.Headers)
 			}
 
+			// The end first: a chunk can still be on its way when the head
+			// has landed, and the end is the frame that says none is.
+			end := decode[appproto.APIEnd](t, rig.frames.await(t, appproto.MsgAPIEnd))
 			var out []byte
 			for _, env := range rig.frames.snapshot() {
 				if env.T == appproto.MsgAPIChunk {
 					out = append(out, decode[appproto.APIChunk](t, env).Data...)
 				}
 			}
-			rig.frames.await(t, appproto.MsgAPIEnd)
+			if end.Truncated || end.Bytes != int64(len(out)) {
+				t.Fatalf("end = %+v, but %d bytes arrived", end, len(out))
+			}
 
 			var answer struct {
 				Modes []struct {

go/internal/api/api_viewer_writes_test.go

@@ -208,11 +208,17 @@ func TestAViewerMayStillRead(t *testing.T) {
 			Method: appproto.APIGet, Path: path,
 		})
 		env := awaitID(t, rig.frames, id)
+		if env.T == appproto.MsgError {
+			t.Fatalf("a viewer's read of %s was refused: %+v", path, decode[appproto.ErrorBody](t, env))
+		}
 		if env.T != appproto.MsgAPIHead {
 			t.Fatalf("a viewer's read of %s answered %s, want a head", path, env.T)
 		}
 		if head := decode[appproto.APIHeadMsg](t, env); head.Status != http.StatusOK {
 			t.Fatalf("a viewer's read of %s answered %d, want 200", path, head.Status)
 		}
+		// The box serves one call at a time, and the app waits for each to
+		// end before it asks again. Asking sooner would be refused as busy.
+		awaitIDType(t, rig.frames, id, appproto.MsgAPIEnd)
 	}
 }

go/internal/appproto/harness_test.go

@@ -342,6 +342,14 @@ func newRigWith(t *testing.T, prefs PlannerPrefs) (*Handler, *fakeBox, *recorder
 
 // deliver hands one client message to the handler as bytes.
 func deliver(t *testing.T, h *Handler, msgType string, id *uint32, b any) {
+	t.Helper()
+	if err := h.Handle(context.Background(), clientFrame(t, msgType, id, b)); err != nil {
+		t.Fatalf("handle %s: %v", msgType, err)
+	}
+}
+
+// clientFrame is one client message as the bytes deliver hands over.
+func clientFrame(t *testing.T, msgType string, id *uint32, b any) []byte {
 	t.Helper()
 	env, err := newEnvelope(msgType, id, b)
 	if err != nil {
@@ -359,9 +367,7 @@ func deliver(t *testing.T, h *Handler, msgType string, id *uint32, b any) {
 	if err != nil {
 		t.Fatalf("frame %s: %v", msgType, err)
 	}
-	if err := h.Handle(context.Background(), frame); err != nil {
-		t.Fatalf("handle %s: %v", msgType, err)
-	}
+	return frame
 }
 
 func subscribe(t *testing.T, h *Handler, rec *recorder) {

go/internal/appproto/passthrough.go

@@ -31,6 +31,7 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/srcfl/ftw/go/internal/apiauth"
@@ -156,10 +157,19 @@ func (h *Handler) onAPIReq(env Envelope) error {
 		})
 	}
 
+	// The slot comes back just before the answer's last frame goes out, not
+	// after it. The app asks again the moment an answer ends; freed after the
+	// frame, the box could refuse that as busy, and the view that lost the
+	// race would wait out a backoff. Any other path (a revoke, a failed send,
+	// an answer refused while its handler still runs) frees the slot when
+	// this goroutine ends, and the Once stops that from freeing a later
+	// request's slot.
 	id := *env.ID
 	go func() {
-		defer h.apiBusy.Store(false)
-		if err := h.serveAPI(id, req); err != nil {
+		var once sync.Once
+		free := func() { once.Do(func() { h.apiBusy.Store(false) }) }
+		defer free()
+		if err := h.serveAPI(id, req, free); err != nil {
 			h.log.Warn("api passthrough failed", "err", err)
 		}
 	}()
@@ -200,7 +210,9 @@ func badAPIRequest(req APIReq) string {
 // Serving it
 // --------------------------------------------------------------------------
 
-func (h *Handler) serveAPI(id uint32, req APIReq) error {
+// free hands the one-at-a-time slot back. serveAPI calls it just before the
+// last frame of an answer, and only once the handler has returned.
+func (h *Handler) serveAPI(id uint32, req APIReq, free func()) error {
 	ctx, cancel := context.WithTimeout(h.ctx, apiHandlerTimeout)
 	defer cancel()
 
@@ -225,6 +237,7 @@ func (h *Handler) serveAPI(id uint32, req APIReq) error {
 		h.log.Info("api passthrough refused",
 			"subject", caller.Subject, "method", req.Method, "path", req.Path,
 			"tier", facts.Tier, "code", refusal.Code)
+		free()
 		return h.sendError(&id, *refusal)
 	}
 
@@ -241,7 +254,7 @@ func (h *Handler) serveAPI(id uint32, req APIReq) error {
 		max = req.MaxBytes
 	}
 	writer := &apiWriter{
-		h: h, id: id, ctx: ctx, max: max, header: http.Header{},
+		h: h, id: id, ctx: ctx, max: max, free: free, header: http.Header{},
 		buf: make([]byte, 0, APIChunkBytes),
 	}
 	h.serveHandler(writer, request)
@@ -457,10 +470,11 @@ func buildAPIRequest(ctx context.Context, caller apiauth.Caller, req APIReq) *ht
 
 // apiWriter turns an http.ResponseWriter into api.head, api.chunk and api.end.
 type apiWriter struct {
-	h   *Handler
-	id  uint32
-	ctx context.Context
-	max int64
+	h    *Handler
+	id   uint32
+	ctx  context.Context
+	max  int64
+	free func()
 
 	header    http.Header
 	buf       []byte
@@ -595,6 +609,7 @@ func (w *apiWriter) finish() error {
 			if w.panicked {
 				reason = "handler"
 			}
+			w.free()
 			return w.h.sendError(&w.id, ErrorBody{
 				Code:      ErrUnavailable,
 				Retryable: ErrorRetryable[ErrUnavailable],
@@ -618,6 +633,7 @@ func (w *apiWriter) finish() error {
 	// middle. The app is committed to the status by then, so the only honest
 	// thing left to say is that this is not the whole of it.
 	truncated := w.truncated || w.ctx.Err() != nil || w.panicked
+	w.free()
 	return w.h.sendBulk(MsgAPIEnd, &w.id, APIEnd{Bytes: w.written, Truncated: truncated})
 }
 

go/internal/appproto/passthrough_test.go

@@ -8,6 +8,7 @@ import (
 	"net/http"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -516,6 +517,82 @@ func TestASecondRequestWhileOneIsInFlightIsRefused(t *testing.T) {
 	}
 }
 
+// askAgainAtEnd is a phone with no delay at all. It hands the box its next
+// request from inside the Send that carries the last frame of the answer
+// before.
+type askAgainAtEnd struct {
+	*recorder
+	h      *Handler
+	last   string
+	next   []byte
+	asked  atomic.Bool
+	handed chan error
+}
+
+func (s *askAgainAtEnd) Send(frame []byte) error {
+	if err := s.recorder.Send(frame); err != nil {
+		return err
+	}
+	f, err := testCodec{}.DecodeFrame(frame)
+	if err == nil && f.Envelope.T == s.last && s.asked.CompareAndSwap(false, true) {
+		s.handed <- s.h.Handle(context.Background(), s.next)
+	}
+	return nil
+}
+
+// The app waits for the end of one answer and then asks again. The box used
+// to free its one slot only after that last frame had gone, so a phone quick
+// enough was told busy about a call that had already ended.
+func TestTheNextRequestIsServedOnceAnAnswerEnds(t *testing.T) {
+	for _, c := range []struct {
+		name string
+		tier apiauth.Tier
+		last string
+	}{
+		{"after an answer", apiauth.TierRead, MsgAPIEnd},
+		{"after a refusal", apiauth.TierLocal, MsgError},
+	} {
+		t.Run(c.name, func(t *testing.T) {
+			h, _, rec, _ := newAPIRig(t, &stubAPI{
+				facts: apiauth.RouteFacts{Tier: c.tier},
+				serve: text(`{}`),
+			})
+			sender := &askAgainAtEnd{
+				recorder: rec, h: h, last: c.last, handed: make(chan error, 1),
+				next: clientFrame(t, MsgAPIReq, ptrU32(41), APIReq{Method: APIGet, Path: "/api/status"}),
+			}
+			h.cfg.Sender = sender
+
+			deliver(t, h, MsgAPIReq, ptrU32(40), APIReq{Method: APIGet, Path: "/api/status"})
+			select {
+			case err := <-sender.handed:
+				if err != nil {
+					t.Fatalf("the next request: %v", err)
+				}
+			case <-time.After(2 * time.Second):
+				t.Fatalf("the first answer never ended; got %s", rec.types())
+			}
+
+			deadline := time.Now().Add(2 * time.Second)
+			for time.Now().Before(deadline) {
+				for _, f := range rec.snapshot() {
+					if f.env.ID == nil || *f.env.ID != 41 {
+						continue
+					}
+					if f.env.T == MsgError {
+						if err := body[ErrorBody](t, f); err.Code == ErrUnavailable {
+							t.Fatalf("the next request met %+v the moment the answer before it ended", err)
+						}
+					}
+					return
+				}
+				time.Sleep(time.Millisecond)
+			}
+			t.Fatalf("no answer to the next request; got %s", rec.types())
+		})
+	}
+}
+
 // --------------------------------------------------------------------------
 // Revocation
 // --------------------------------------------------------------------------
@@ -732,22 +809,6 @@ func stepUpClock(t *testing.T, h *Handler) *fakeClock {
 	return c
 }
 
-// waitIdle blocks until the passthrough queue is free, so the next request is
-// served rather than refused as busy. The frame lands just before the serving
-// goroutine clears the flag, so a test that only waited on the frame could race
-// the very next deliver.
-func waitIdle(t *testing.T, h *Handler) {
-	t.Helper()
-	deadline := time.Now().Add(2 * time.Second)
-	for time.Now().Before(deadline) {
-		if !h.apiBusy.Load() {
-			return
-		}
-		time.Sleep(time.Millisecond)
-	}
-	t.Fatal("passthrough stayed busy")
-}
-
 // The security property the window must not weaken: a configure with no
 // ceremony behind it, and no recent one either, is still refused. This is the
 // first write of any session — the app was never meant to skip it.
@@ -819,7 +880,6 @@ func TestANoStepUpWriteDoesNotOpenTheWindow(t *testing.T) {
 		Method: APIPut, Path: "/api/loadpoints/1/schedule",
 	})
 	waitFor(t, rec, MsgAPIEnd)
-	waitIdle(t, h)
 
 	api.facts = apiauth.RouteFacts{Tier: apiauth.TierConfigure}
 	rec.reset()
@@ -853,7 +913,6 @@ func TestASecondConfigureWithinTheWindowSkipsTheCeremony(t *testing.T) {
 		t.Fatalf("the ceremony write answered %d, want 200", head.Status)
 	}
 	waitFor(t, rec, MsgAPIEnd)
-	waitIdle(t, h)
 	rec.reset()
 
 	// A later write in the same screen, four minutes on, carrying no ceremony.
@@ -883,7 +942,6 @@ func TestAConfigureAfterTheWindowNeedsAFreshCeremony(t *testing.T) {
 		Method: APIPost, Path: "/api/battery_covers_ev", StepUp: true,
 	})
 	waitFor(t, rec, MsgAPIEnd)
-	waitIdle(t, h)
 	rec.reset()
 
 	clock.uptimeMs += StepUpWindowMs