amadan

oficina / docs Public

Prove stream lease deadlines across real session writer contention

Paul Campbell pushed by paul@keymail.dev 2a5de8b3d27490a277304292f89fa57474f86097 parent 7cf6417
1 file changed, +119 −0
  • internal/docs/team_stream_test.go +119 −0

CI passed — run details

diff --git a/internal/docs/team_stream_test.go b/internal/docs/team_stream_test.go
new file mode 100644
index 0000000..d20255a
--- /dev/null
+++ b/internal/docs/team_stream_test.go
@@ -0,0 +1,119 @@
+package docs
+
+import (
+ "amadan.net/oficina/home/roster"
+ "bufio"
+ "context"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "testing"
+ "time"
+)
+
+func teamOpenActualStream(t *testing.T, f *teamRouterFixture, b *routerBrowser, srv *httptest.Server) *http.Response {
+ t.Helper()
+ r, e := http.NewRequestWithContext(t.Context(), "GET", srv.URL+"/d/"+f.own.Sqid+"/stream", nil)
+ if e != nil {
+ t.Fatal(e)
+ }
+ r.Host = strings.TrimPrefix(bridgeOrigin, "https://")
+ for _, c := range b.cookies {
+ r.AddCookie(c)
+ }
+ res, e := srv.Client().Do(r)
+ if e != nil {
+ t.Fatal(e)
+ }
+ t.Cleanup(func() { res.Body.Close() })
+ if res.StatusCode != 200 {
+ raw, _ := io.ReadAll(res.Body)
+ t.Fatal("stream refused", res.StatusCode, res.Request.URL.String(), string(raw))
+ }
+ reader := bufio.NewReader(res.Body)
+ name, _ := nextEvent(t, reader, time.Second)
+ if name != "update" {
+ t.Fatal("initial sequence event missing")
+ }
+ // Keep the same buffered reader: it may already contain the next frame.
+ res.Body = struct {
+ io.Reader
+ io.Closer
+ }{reader, res.Body}
+ return res
+}
+func TestTeamStreamActualSessionWriterWaitCannotExtendOriginalLease(t *testing.T) {
+ f := newTeamRouterFixture(t)
+ b := f.browser()
+ b.localSignin(t, "alice@example.com")
+ b.linkProof(t, teamSubject('a'), "alice@example.com")
+ routerRedirect(t, b.do("POST", "/account/home/confirm", ""), "/")
+ if r := b.do("GET", "/d/"+f.own.Sqid, ""); r.Code != 200 {
+ t.Fatal("initial membership unavailable", r.Code)
+ }
+ streamDurations(t, f.app, 40*time.Millisecond, 5*time.Second)
+ f.directory.change(nil, true)
+ f.clock.step(59 * time.Second)
+ if e := f.app.store.team.cache.Refresh(t.Context()); e == nil {
+ t.Fatal("failed directory refresh accepted")
+ }
+ srv := httptest.NewServer(f.app.Handler())
+ defer srv.Close()
+ res := teamOpenActualStream(t, f, b, srv)
+ defer res.Body.Close()
+ writer := f.app.db.Writer()
+ tx, e := writer.BeginTx(t.Context(), nil)
+ if e != nil {
+ t.Fatal(e)
+ }
+ defer tx.Rollback()
+ baseline := writer.Stats().WaitCount
+ done := make(chan struct{})
+ go func() { io.Copy(io.Discard, res.Body); close(done) }()
+ deadline := time.Now().Add(300 * time.Millisecond)
+ for writer.Stats().WaitCount == baseline && time.Now().Before(deadline) {
+ time.Sleep(time.Millisecond)
+ }
+ if writer.Stats().WaitCount == baseline {
+ t.Fatal("actual session reader never queued behind writer")
+ }
+ select {
+ case <-done:
+ case <-time.After(1300 * time.Millisecond):
+ t.Fatal("writer wait held stream beyond its original lease")
+ }
+ // The stream ended while the writer remains locked; cancellation did not need
+ // the transaction to finish and did not mint a new session.
+ if e = tx.Rollback(); e != nil {
+ t.Fatal(e)
+ }
+ if _, ok := f.sessions.From(b.request("GET", "/", "")); !ok {
+ t.Fatal("stream cancellation revoked legacy session")
+ }
+}
+func TestTeamStreamHealthyPublicationClosesOriginalLeaseAndReconnects(t *testing.T) {
+ f := newTeamRouterFixture(t)
+ b := f.browser()
+ b.localSignin(t, "alice@example.com")
+ b.linkProof(t, teamSubject('a'), "alice@example.com")
+ routerRedirect(t, b.do("POST", "/account/home/confirm", ""), "/")
+ streamDurations(t, f.app, 30*time.Millisecond, 5*time.Second)
+ srv := httptest.NewServer(f.app.Handler())
+ defer srv.Close()
+ res := teamOpenActualStream(t, f, b, srv)
+ defer res.Body.Close()
+ f.clock.step(roster.RetryAfter)
+ if e := f.app.store.team.cache.Refresh(context.Background()); e != nil {
+ t.Fatal(e)
+ }
+ done := make(chan struct{})
+ go func() { io.Copy(io.Discard, res.Body); close(done) }()
+ select {
+ case <-done:
+ case <-time.After(time.Second):
+ t.Fatal("old stream renewed instead of closing after publication")
+ }
+ fresh := teamOpenActualStream(t, f, b, srv)
+ fresh.Body.Close()
+}