diff --git a/.changeset/api-passthrough-free-before-end.md b/.changeset/api-passthrough-free-before-end.md new file mode 100644 index 000000000..36ce9925c --- /dev/null +++ b/.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. diff --git a/go/internal/api/api_passthrough_test.go b/go/internal/api/api_passthrough_test.go index 37651f330..b893f9b19 100644 --- a/go/internal/api/api_passthrough_test.go +++ b/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 { diff --git a/go/internal/api/api_viewer_writes_test.go b/go/internal/api/api_viewer_writes_test.go index 9ff91e725..0477003c9 100644 --- a/go/internal/api/api_viewer_writes_test.go +++ b/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) } } diff --git a/go/internal/appproto/harness_test.go b/go/internal/appproto/harness_test.go index 24ac371c1..c193c8d78 100644 --- a/go/internal/appproto/harness_test.go +++ b/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) { diff --git a/go/internal/appproto/passthrough.go b/go/internal/appproto/passthrough.go index 608b7b989..a79c4e771 100644 --- a/go/internal/appproto/passthrough.go +++ b/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}) } diff --git a/go/internal/appproto/passthrough_test.go b/go/internal/appproto/passthrough_test.go index 8c982f38a..c34f105b7 100644 --- a/go/internal/appproto/passthrough_test.go +++ b/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