diff --git a/.changeset/relay-rotation-gap.md b/.changeset/relay-rotation-gap.md new file mode 100644 index 000000000..bb1c6d64e --- /dev/null +++ b/.changeset/relay-rotation-gap.md @@ -0,0 +1,5 @@ +--- +"ftw": patch +--- + +A phone that opens the app just after the hour finds the box at once over the relay. The relay turns its rendezvous handles on the hour but closes each box's old room on its own turn in a five-minute spread, so a phone could wait up to five minutes in an empty room. The box now joins the new hour's handle when the hour turns and keeps the old one until the relay closes it. diff --git a/go/internal/appuplink/client.go b/go/internal/appuplink/client.go index 43f6ed93a..7571605d5 100644 --- a/go/internal/appuplink/client.go +++ b/go/internal/appuplink/client.go @@ -25,6 +25,7 @@ import ( "net/http" "net/url" "regexp" + "slices" "strconv" "strings" "sync" @@ -71,16 +72,29 @@ const ( // backoffCap keeps a box off a struggling relay without letting it sulk. backoffCap = 60 * time.Second - // rotateJitter is how long a box waits before rejoining after an epoch - // turned over. + // rotateJitter is the longest a box waits before it joins a new epoch's + // handle: after its clock turns the hour, or after the relay closed the + // old handle. // - // Not zero. The relay closes the old handle and the new one appears when - // the peer comes back; returning instantly gives it a stronger timing link - // between the two protocol handles. The relay already spreads the closes - // across five minutes, so a few seconds here weakens that signal without - // making the phone wait. It cannot hide IP or connection continuity. + // Not zero. Every box turns at the same hour, so this spreads their + // joins, and coming back the instant the relay closes the old handle + // gives it a stronger timing link between the two protocol handles. It + // cannot hide IP or connection continuity. rotateJitter = 5 * time.Second + // turnRetry is how long a box waits to try the new hour's handle again + // when the relay refused it with the old epoch. The relay reads its + // clock on its fifteen-second heartbeat, so it can still be on the old + // hour that long after ours turned; turnTries tries cover that. After + // them the box waits for the relay to close the old room, as it always + // has. + turnRetry = 5 * time.Second + turnTries = 4 + + // turnCheck is how often a connected box looks at its clock for a new + // hour. + turnCheck = time.Second + // correctionLimit is how many epoch corrections we take at full speed // before backing off. More than a couple means something is wrong rather // than stale. @@ -157,14 +171,13 @@ type Uplink struct { attempt int corrections int - // liveMu guards live, the sessions of the current relay connection, - // and liveText, the same connection's text-frame writer (the deadman - // claims ride on it). Kept on the uplink so a revoke can reach into - // whatever connection is up right now; nil between connections, when - // there is nothing to drop and nothing to say. - liveMu sync.Mutex - live *sessions - liveText func(string) error + // liveMu guards live, the relay connections up right now, oldest first. + // Usually one; two for the few minutes after the hour while the box + // holds the old handle and the new. Kept on the uplink so a revoke can + // reach whatever is up right now; empty between connections, when there + // is nothing to drop and nothing to say. + liveMu sync.Mutex + live []*liveLink // The dead man's switch bookkeeping: which row ids the relay holds, // so a removed subscription's row is deleted rather than left armed. @@ -174,18 +187,28 @@ type Uplink struct { deadmanHTTP *http.Client deadmanRefresh time.Duration readTimeout time.Duration + turnCheck time.Duration + turnRetry time.Duration +} + +// liveLink is one relay connection's sessions and its text-frame writer, +// which the deadman claims ride on. +type liveLink struct { + sessions *sessions + text func(string) error } // DropSessionsByAppKey tears down every live session a revoked key holds. // Returns how many went; zero is normal when the phone is not connected. func (u *Uplink) DropSessionsByAppKey(pub []byte) int { u.liveMu.Lock() - live := u.live + live := slices.Clone(u.live) u.liveMu.Unlock() - if live == nil { - return 0 + dropped := 0 + for _, l := range live { + dropped += l.sessions.dropByAppKey(pub) } - return live.dropByAppKey(pub) + return dropped } // New builds an uplink. @@ -225,6 +248,8 @@ func New(opts Options) (*Uplink, error) { deadmanPosted: map[string]bool{}, deadmanRefresh: opts.DeadmanRefresh, readTimeout: defaultReadTimeout, + turnCheck: turnCheck, + turnRetry: turnRetry, } if u.dialer == nil { u.dialer = productionDialer() @@ -284,7 +309,7 @@ func (u *Uplink) Run(ctx context.Context) error { // once dials, serves until the connection ends, and says how long to wait. func (u *Uplink) once(ctx context.Context) (time.Duration, error) { - epoch := CurrentEpoch(u.now().UnixMilli()) + u.epochOffset + epoch := u.epoch() handle, err := Handle(u.opts.Enroll.RendezvousSecret(), epoch) if err != nil { // Unrecoverable and not the network's fault. Back off rather than @@ -292,14 +317,7 @@ func (u *Uplink) once(ctx context.Context) (time.Duration, error) { return backoffCap, err } - target := fmt.Sprintf("%s/r/%d/%s/box", u.opts.Endpoint, epoch, handle) - - dialCtx, cancel := context.WithTimeout(ctx, dialTimeout) - defer cancel() - conn, response, err := u.dialer.DialContext(dialCtx, target, nil) - if response != nil && response.Body != nil { - _ = response.Body.Close() - } + conn, err := u.dial(ctx, epoch, handle) if err != nil { return u.backoff(), fmt.Errorf("appuplink: dialling the relay: %w", err) } @@ -308,20 +326,20 @@ func (u *Uplink) once(ctx context.Context) (time.Duration, error) { // this, an hourly rotation would compound the backoff until a box that // works perfectly waits a minute after every rotation. connectedAt := u.now() - code, reason, serveErr := u.serve(ctx, conn) + end := u.ride(ctx, u.start(ctx, conn, epoch)) if u.now().Sub(connectedAt) >= backoffCap { u.attempt = 0 } - switch code { + switch end.code { case CloseRotated: - u.adoptEpoch(reason) + u.adoptEpoch(end.reason) u.attempt = 0 u.corrections = 0 - u.log.Info("relay epoch rotated", "epoch", CurrentEpoch(u.now().UnixMilli())+u.epochOffset) + u.log.Info("relay epoch rotated", "epoch", u.epoch()) return time.Duration(u.random() * float64(rotateJitter)), nil case CloseEpoch: - u.adoptEpoch(reason) + u.adoptEpoch(end.reason) u.corrections++ if u.corrections > correctionLimit { return u.backoff(), fmt.Errorf("appuplink: the relay keeps refusing our epoch") @@ -330,8 +348,164 @@ func (u *Uplink) once(ctx context.Context) (time.Duration, error) { // right after one round trip. return 0, nil default: - return u.backoff(), serveErr + return u.backoff(), end.err + } +} + +// epoch is the epoch the box would join now: its clock's guess, corrected by +// what the relay last told it. +func (u *Uplink) epoch() int64 { + return CurrentEpoch(u.now().UnixMilli()) + u.epochOffset +} + +// dial opens a socket under one epoch's handle. The relay answers a wrong +// epoch after the upgrade, with a close, so an open socket is not yet an +// accepted one. +func (u *Uplink) dial(ctx context.Context, epoch int64, handle string) (*websocket.Conn, error) { + target := fmt.Sprintf("%s/r/%d/%s/box", u.opts.Endpoint, epoch, handle) + + dialCtx, cancel := context.WithTimeout(ctx, dialTimeout) + defer cancel() + conn, response, err := u.dialer.DialContext(dialCtx, target, nil) + if response != nil && response.Body != nil { + _ = response.Body.Close() + } + return conn, err +} + +// link is one relay connection, served on its own goroutine. +type link struct { + epoch int64 + stop context.CancelFunc + done chan linkEnd +} + +// linkEnd is how a connection ended: the relay's close code and reason, or +// the error when it ended some other way. +type linkEnd struct { + code int + reason string + err error +} + +func (u *Uplink) start(ctx context.Context, conn *websocket.Conn, epoch int64) *link { + ctx, stop := context.WithCancel(ctx) + l := &link{epoch: epoch, stop: stop, done: make(chan linkEnd, 1)} + go func() { + defer stop() + code, reason, err := u.serve(ctx, conn) + l.done <- linkEnd{code: code, reason: reason, err: err} + }() + return l +} + +// end stops a connection and waits until its sessions are closed. +func (l *link) end() { + if l == nil { + return + } + l.stop() + <-l.done +} + +// doneOf is l.done, or nil, which never fires in a select, when there is no +// connection. +func doneOf(l *link) chan linkEnd { + if l == nil { + return nil + } + return l.done +} + +// ride serves cur until it ends, and joins each new hour's handle beside it. +// +// The relay turns its epoch on the hour and from then on refuses the old +// one, but it closes each old room only on that room's turn in a five-minute +// spread. A box that moved only when closed left every phone that dialled in +// between waiting in an empty room for up to five minutes; on 13 Oct 2026 a +// phone waited from 09:00 until the relay closed its box's room at 09:03:18. +// So when the hour turns the box also joins the new handle, and keeps the +// old one until the relay closes it: a phone already talking to the box keeps +// it, and a phone that dials the new handle finds it. +// +// It returns how the last connection ended, once no other is up to carry on. +func (u *Uplink) ride(ctx context.Context, cur *link) linkEnd { + // Early joins live only as long as this call, so none outlives it. + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + var ( + next *link + joining chan *link + tries int + // tryAt is on the real clock, like every other delay here. Zero + // until this hour's turn has been seen. + tryAt time.Time + ) + check := time.NewTicker(u.turnCheck) + defer check.Stop() + + for { + select { + case end := <-cur.done: + if next == nil || ctx.Err() != nil { + next.end() + return end + } + if end.code == CloseRotated { + u.adoptEpoch(end.reason) + } + cur, next = next, nil + tries, tryAt = 0, time.Time{} + u.attempt, u.corrections = 0, 0 + u.log.Info("relay epoch rotated", "epoch", cur.epoch, "joined_early", true) + + case <-doneOf(next): + // The new handle did not hold. Most often the relay has not + // read its clock since the hour and still says the old epoch, + // which is not a correction to take. The old room carries on. + next = nil + tries++ + tryAt = time.Now().Add(u.turnRetry) + + case l := <-joining: + joining = nil + if l == nil { + tries++ + tryAt = time.Now().Add(u.turnRetry) + continue + } + next = l + + case <-check.C: + want := u.epoch() + if next != nil || joining != nil || tries >= turnTries || want <= cur.epoch { + continue + } + if tryAt.IsZero() { + tryAt = time.Now().Add(time.Duration(u.random() * float64(rotateJitter))) + } + if time.Now().Before(tryAt) { + continue + } + joining = make(chan *link, 1) + go func(result chan<- *link) { result <- u.join(ctx, want) }(joining) + } + } +} + +// join opens the new hour's handle beside the old one. Nil is a failed try. +func (u *Uplink) join(ctx context.Context, epoch int64) *link { + handle, err := Handle(u.opts.Enroll.RendezvousSecret(), epoch) + if err != nil { + return nil + } + conn, err := u.dial(ctx, epoch, handle) + if err != nil { + u.log.Debug("could not join the new hour's handle yet", "err", err) + return nil } + return u.start(ctx, conn, epoch) } // serve runs one connection. It returns the close code and reason when the @@ -371,14 +545,13 @@ func (u *Uplink) serve(ctx context.Context, conn *websocket.Conn) (int, string, write: write, log: u.log, } + here := &liveLink{sessions: live, text: writeText} u.liveMu.Lock() - u.live = live - u.liveText = writeText + u.live = append(u.live, here) u.liveMu.Unlock() defer func() { u.liveMu.Lock() - u.live = nil - u.liveText = nil + u.live = slices.DeleteFunc(u.live, func(l *liveLink) bool { return l == here }) u.liveMu.Unlock() live.closeAll() }() diff --git a/go/internal/appuplink/deadman.go b/go/internal/appuplink/deadman.go index c4043aaf0..9e2eca608 100644 --- a/go/internal/appuplink/deadman.go +++ b/go/internal/appuplink/deadman.go @@ -161,11 +161,16 @@ func (u *Uplink) ResyncDeadman() { u.deadmanPosted = posted u.deadmanMu.Unlock() - // Claim what was posted, if there is a socket to claim on. The claim - // is per-connection state on the relay, so serve() runs this again on - // every reconnect. + // Claim what was posted on the newest socket, if there is one. The + // claim is per-connection state on the relay, so serve() runs this again + // on every connect. In the overlap after the hour the older socket keeps + // the claims it made on its own connect, and the relay counts claims, so + // closing one socket does not release the other's. u.liveMu.Lock() - text := u.liveText + var text func(string) error + if n := len(u.live); n > 0 { + text = u.live[n-1].text + } u.liveMu.Unlock() if text == nil { return diff --git a/go/internal/appuplink/rotation_test.go b/go/internal/appuplink/rotation_test.go new file mode 100644 index 000000000..68eed1239 --- /dev/null +++ b/go/internal/appuplink/rotation_test.go @@ -0,0 +1,450 @@ +package appuplink + +import ( + "context" + "log/slog" + "net/http" + "net/http/httptest" + "path/filepath" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/gorilla/websocket" + "github.com/srcfl/ftw/go/internal/apiauth" + "github.com/srcfl/ftw/go/internal/appenroll" + "github.com/srcfl/ftw/go/internal/appproto" +) + +// roomRelay is the part of srcfl/ftw-webapp relay/src/server.ts that +// fakeRelay leaves out: a room per handle, an exact epoch on every join, and a +// rotation that closes each old room with 4410 on its own turn. +// +// The test is the relay's clock. turn() is the heartbeat that notices the +// hour; until then the relay still refuses the new epoch. rotate() is one +// room's turn in the five-minute spread; until then the old room stays open. +type roomRelay struct { + server *httptest.Server + + mu sync.Mutex + epoch int64 + rooms map[string]*relayRoom + // refused counts joins answered with 4409. + refused int +} + +type relayRoom struct { + epoch int64 + uplink *relayPeer + streams []*relayPeer +} + +type relayPeer struct { + conn *websocket.Conn + // writeMu because gorilla allows one writer at a time, and the relay + // writes to a peer from the other peers' read loops. + writeMu sync.Mutex + linked bool +} + +func (p *relayPeer) write(kind int, message []byte) { + p.writeMu.Lock() + defer p.writeMu.Unlock() + _ = p.conn.WriteMessage(kind, message) +} + +func (p *relayPeer) close(code int, reason string) { + p.writeMu.Lock() + defer p.writeMu.Unlock() + _ = p.conn.WriteControl(websocket.CloseMessage, + websocket.FormatCloseMessage(code, reason), time.Now().Add(time.Second)) + _ = p.conn.Close() +} + +func newRoomRelay(t *testing.T, epoch int64) *roomRelay { + t.Helper() + r := &roomRelay{epoch: epoch, rooms: map[string]*relayRoom{}} + r.server = httptest.NewServer(http.HandlerFunc(r.handle)) + t.Cleanup(r.server.Close) + return r +} + +func (r *roomRelay) url() string { + return "ws" + strings.TrimPrefix(r.server.URL, "http") +} + +// turn is the relay's heartbeat noticing that the hour has changed. +func (r *roomRelay) turn(epoch int64) { + r.mu.Lock() + r.epoch = epoch + r.mu.Unlock() +} + +// rotate is one old room's turn in the spread: everyone in it hears 4410 and +// the new epoch, and the room is gone. +func (r *roomRelay) rotate(handle string) { + r.mu.Lock() + room := r.rooms[handle] + if room == nil || room.epoch >= r.epoch { + r.mu.Unlock() + return + } + delete(r.rooms, handle) + peers := append([]*relayPeer(nil), room.streams...) + if room.uplink != nil { + peers = append(peers, room.uplink) + } + epoch := r.epoch + r.mu.Unlock() + + for _, p := range peers { + p.close(CloseRotated, strconv.FormatInt(epoch, 10)) + } +} + +func (r *roomRelay) refusals() int { + r.mu.Lock() + defer r.mu.Unlock() + return r.refused +} + +func (r *roomRelay) hasUplink(handle string) bool { + r.mu.Lock() + defer r.mu.Unlock() + room := r.rooms[handle] + return room != nil && room.uplink != nil +} + +func (r *roomRelay) handle(w http.ResponseWriter, req *http.Request) { + if strings.HasPrefix(req.URL.Path, "/deadman") { + w.WriteHeader(http.StatusNoContent) + return + } + parts := strings.Split(strings.TrimPrefix(req.URL.Path, "/"), "/") + if len(parts) != 4 || parts[0] != "r" { + http.Error(w, "join", http.StatusBadRequest) + return + } + epoch, err := strconv.ParseInt(parts[1], 10, 64) + if err != nil { + http.Error(w, "join", http.StatusBadRequest) + return + } + handle, role := parts[2], parts[3] + + conn, err := (&websocket.Upgrader{}).Upgrade(w, req, nil) + if err != nil { + return + } + peer := &relayPeer{conn: conn} + + r.mu.Lock() + if epoch != r.epoch { + // Exact, as server.ts is: the old epoch is refused the moment the + // relay has turned, even while old rooms are still open. + want := r.epoch + r.refused++ + r.mu.Unlock() + peer.close(CloseEpoch, strconv.FormatInt(want, 10)) + return + } + room := r.rooms[handle] + if room == nil { + room = &relayRoom{epoch: epoch} + r.rooms[handle] = room + } + if role == "box" { + if room.uplink != nil { + r.mu.Unlock() + peer.close(CloseBusy, "uplink") + return + } + room.uplink = peer + } else { + room.streams = append(room.streams, peer) + } + r.sync(room) + r.mu.Unlock() + + defer func() { + r.mu.Lock() + if room.uplink == peer { + room.uplink = nil + } + for i, s := range room.streams { + if s == peer { + room.streams = append(room.streams[:i], room.streams[i+1:]...) + break + } + } + if room.uplink == nil && len(room.streams) == 0 { + if r.rooms[handle] == room { + delete(r.rooms, handle) + } + } else { + r.sync(room) + } + r.mu.Unlock() + _ = conn.Close() + }() + + for { + kind, message, err := conn.ReadMessage() + if err != nil { + return + } + if kind != websocket.BinaryMessage { + if role == "box" && strings.HasPrefix(string(message), CtrlDeadman+" ") { + continue + } + return + } + r.mu.Lock() + var to []*relayPeer + if room.uplink == peer { + to = append(to, room.streams...) + } else if room.uplink != nil { + to = append(to, room.uplink) + } + r.mu.Unlock() + for _, p := range to { + p.write(websocket.BinaryMessage, message) + } + } +} + +// sync tells each side whether the other is there, as server.ts does. Called +// with r.mu held. +func (r *roomRelay) sync(room *relayRoom) { + notify := func(p *relayPeer, present bool) { + if p.linked == present { + return + } + p.linked = present + word := CtrlGone + if present { + word = CtrlReady + } + go p.write(websocket.TextMessage, []byte(word)) + } + if room.uplink != nil { + notify(room.uplink, len(room.streams) > 0) + } + for _, s := range room.streams { + notify(s, room.uplink != nil) + } +} + +// manualClock is the box's wall clock, moved by the test. +type manualClock struct { + mu sync.Mutex + t time.Time +} + +func (c *manualClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.t +} + +func (c *manualClock) Set(t time.Time) { + c.mu.Lock() + c.t = t + c.mu.Unlock() +} + +type rotationRig struct { + relay *roomRelay + clock *manualClock + enroll *appenroll.Identity + uplink *Uplink +} + +// newRotationRig starts a box whose clock reads one minute before the end of +// epoch, on a relay that is on epoch. +func newRotationRig(t *testing.T, epoch int64) *rotationRig { + t.Helper() + relay := newRoomRelay(t, epoch) + clock := &manualClock{t: time.UnixMilli((epoch+1)*EpochMs - 60_000)} + + enroll, err := appenroll.LoadOrCreate(filepath.Join(t.TempDir(), "nova.key")) + if err != nil { + t.Fatalf("LoadOrCreate: %v", err) + } + u, err := New(Options{ + Endpoint: relay.url(), + Enroll: enroll, + Handler: newHandler, + Logger: slog.New(slog.DiscardHandler), + Now: clock.Now, + Random: func() float64 { return 0 }, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + u.turnCheck = 5 * time.Millisecond + u.turnRetry = 100 * time.Millisecond + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- u.Run(ctx) }() + t.Cleanup(func() { + cancel() + <-done + }) + + r := &rotationRig{relay: relay, clock: clock, enroll: enroll, uplink: u} + r.waitForUplink(t, epoch) + return r +} + +func (r *rotationRig) handle(t *testing.T, epoch int64) string { + t.Helper() + h, err := Handle(r.enroll.RendezvousSecret(), epoch) + if err != nil { + t.Fatalf("Handle: %v", err) + } + return h +} + +func (r *rotationRig) waitForUplink(t *testing.T, epoch int64) { + t.Helper() + handle := r.handle(t, epoch) + deadline := time.Now().Add(5 * time.Second) + for !r.relay.hasUplink(handle) { + if time.Now().After(deadline) { + t.Fatalf("the box never joined epoch %d", epoch) + } + time.Sleep(5 * time.Millisecond) + } +} + +// openSession joins epoch's room as a phone and runs a handshake and a hello. +// It reports whether the box answered within wait. +func (r *rotationRig) openSession(t *testing.T, epoch int64, wait time.Duration) (*websocket.Conn, *appClient, bool) { + t.Helper() + code, _, err := r.enroll.MintPairingCode(apiauth.RoleOwner, appenroll.PairingTTL) + if err != nil { + t.Fatalf("MintPairingCode: %v", err) + } + conn, _, err := websocket.DefaultDialer.Dial( + r.relay.url()+"/r/"+strconv.FormatInt(epoch, 10)+"/"+r.handle(t, epoch)+"/app", nil) + if err != nil { + t.Fatalf("dialling as an app: %v", err) + } + t.Cleanup(func() { _ = conn.Close() }) + + // The app's own rule: wait for "ready", because a frame written into an + // empty room is dropped. + _ = conn.SetReadDeadline(time.Now().Add(wait)) + for { + kind, message, err := conn.ReadMessage() + if err != nil { + return conn, nil, false + } + if kind == websocket.TextMessage && string(message) == CtrlReady { + break + } + } + + app, err := newAppClient() + if err != nil { + t.Fatalf("newAppClient: %v", err) + } + message1, err := app.message1(r.enroll.StaticKey().Public(), code) + if err != nil { + t.Fatalf("message1: %v", err) + } + if err := conn.WriteMessage(websocket.BinaryMessage, message1); err != nil { + t.Fatalf("sending message 1: %v", err) + } + if err := app.readMessage2(waitForBinary(t, conn)); err != nil { + t.Fatalf("readMessage2: %v", err) + } + send(t, conn, app, appproto.MsgHello, 512, appproto.Hello{ + Proto: appproto.ProtoRange{Min: 0, Max: appproto.ProtoMax}, + App: appproto.AppInfo{Build: "test", UA: "ios"}, + }) + if got := receive(t, conn, app); got.Envelope.T != appproto.MsgHelloOK { + t.Fatalf("first reply is %q, want %q", got.Envelope.T, appproto.MsgHelloOK) + } + return conn, app, true +} + +// A phone that opens the app just after the hour derives the new hour's +// handle from its clock. The relay has turned and refuses the old epoch, but +// it closes the box's old room only on that room's turn in its five-minute +// spread. Seen on 13 Oct 2026: the phone waited from 09:00 until the relay +// rotated the box at 09:03:18. The box must already be on the new handle. +func TestAPhoneOpenedAfterTheHourFindsTheBox(t *testing.T) { + const epoch = 497742 + r := newRotationRig(t, epoch) + + // 09:00:30 on both clocks. The old room's turn has not come. + r.clock.Set(time.UnixMilli((epoch+1)*EpochMs + 30_000)) + r.relay.turn(epoch + 1) + + if _, _, ok := r.openSession(t, epoch+1, 3*time.Second); !ok { + t.Fatal("a phone on the new hour's handle found no box until the relay rotated the old room") + } +} + +// A phone already talking to the box must not lose it at the hour. The box +// keeps the old handle until the relay closes that room, and the phone moves +// when the relay tells it to. +func TestAPhoneConnectedAcrossTheHourKeepsTheBox(t *testing.T) { + const epoch = 497742 + r := newRotationRig(t, epoch) + conn, app, ok := r.openSession(t, epoch, 3*time.Second) + if !ok { + t.Fatal("no session before the hour") + } + + r.clock.Set(time.UnixMilli((epoch+1)*EpochMs + 30_000)) + r.relay.turn(epoch + 1) + r.waitForUplink(t, epoch+1) + + // Still the old room, still answered. + send(t, conn, app, appproto.MsgSub, 512, appproto.Sub{Bucket: 512}) + if got := receive(t, conn, app); got.Envelope.T != appproto.MsgSnap { + t.Fatalf("after the hour the old session got %q, want %q", got.Envelope.T, appproto.MsgSnap) + } + + // The room's turn. The phone follows to the new handle and the box is + // there; the box did not have to move. + r.relay.rotate(r.handle(t, epoch)) + if _, _, ok := r.openSession(t, epoch+1, 3*time.Second); !ok { + t.Fatal("after the old room rotated the phone found no box on the new handle") + } + if !r.relay.hasUplink(r.handle(t, epoch+1)) { + t.Fatal("the box left the new handle when the old room rotated") + } +} + +// The relay reads its clock on a fifteen-second heartbeat, so just after the +// hour it may still refuse the new epoch. That is "not yet", not a correction +// to an hour ago: the box keeps the old room and tries again. +func TestTheBoxWaitsForARelayThatHasNotTurned(t *testing.T) { + const epoch = 497742 + r := newRotationRig(t, epoch) + + r.clock.Set(time.UnixMilli((epoch+1)*EpochMs + 1_000)) + deadline := time.Now().Add(5 * time.Second) + for r.relay.refusals() == 0 { + if time.Now().After(deadline) { + t.Fatal("the box never tried the new hour's handle") + } + time.Sleep(2 * time.Millisecond) + } + if !r.relay.hasUplink(r.handle(t, epoch)) { + t.Fatal("a refused try took the box out of the old room") + } + + r.relay.turn(epoch + 1) + r.waitForUplink(t, epoch+1) + if !r.relay.hasUplink(r.handle(t, epoch)) { + t.Fatal("the box left the old room before the relay closed it") + } +}