| 1 | package hub |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "fmt" |
| 6 | "time" |
| 7 | ) |
| 8 | |
| 9 | func isTargeted(t string) bool { return t == TypeOffer || t == TypeAnswer || t == TypeCandidate } |
| 10 | |
| 11 | // Relay delivers one targeted message. The envelope is rebuilt from the |
| 12 | // binding; the payload is never read. |
| 13 | // |
| 14 | // A member's standing is re-derived first (§15.12): the participation |
| 15 | // authorizes everything after entry (§2.4), but a member who has lost |
| 16 | // the team is dropped by the next thing they do, and this is that thing. |
| 17 | // A member with no standing is ended here and the request answers |
| 18 | // ErrNotYours, which the handler renders as the same not-found body an |
| 19 | // unknown peer gets. |
| 20 | func (h *Hub) Relay(ctx context.Context, sqid, session string, from PeerID, m Message) error { |
| 21 | if !isTargeted(m.Type) { |
| 22 | return fmt.Errorf("%w: %q", ErrNotTargeted, m.Type) |
| 23 | } |
| 24 | gone, err := h.memberStandingGone(ctx, sqid, session, from, true) |
| 25 | if err != nil { |
| 26 | return err |
| 27 | } |
| 28 | if gone { |
| 29 | return ErrNotYours |
| 30 | } |
| 31 | h.mu.Lock() |
| 32 | defer h.mu.Unlock() |
| 33 | r, ok := h.rooms[sqid] |
| 34 | if !ok { |
| 35 | return ErrUnknownPeer |
| 36 | } |
| 37 | p, found := h.findLocked(sqid, session, from) |
| 38 | if !found { |
| 39 | return ErrUnknownPeer |
| 40 | } |
| 41 | if !p.Admitted { |
| 42 | return ErrNotAdmitted |
| 43 | } |
| 44 | if m.To == p.Peer { |
| 45 | return ErrUnknownPeer // a peer is not its own target |
| 46 | } |
| 47 | tp, ok := r.peers[m.To] |
| 48 | if !ok || !tp.Admitted { |
| 49 | return ErrUnknownPeer |
| 50 | } |
| 51 | if m.Gen != h.genLocked(r, p.Peer, tp.Peer) { |
| 52 | return ErrStaleGen |
| 53 | } |
| 54 | out := Message{Type: m.Type, From: p.Peer, Name: p.Name, To: tp.Peer, Seq: m.Seq, Gen: m.Gen, Payload: m.Payload} |
| 55 | h.ringLocked(tp, out) |
| 56 | h.sendLocked(tp, out) |
| 57 | return nil |
| 58 | } |
| 59 | |
| 60 | // ringLocked keeps a copy for the resume window, newest last. What falls |
| 61 | // out of the ring is remembered as a high-water mark per sender, which |
| 62 | // is the only way a resume can tell that a sender's messages were |
| 63 | // evicted wholesale by another sender's burst. |
| 64 | func (h *Hub) ringLocked(p *Participation, m Message) { |
| 65 | p.ring = append(p.ring, m) |
| 66 | if len(p.ring) <= RingSize { |
| 67 | return |
| 68 | } |
| 69 | for _, old := range p.ring[:len(p.ring)-RingSize] { |
| 70 | if !isTargeted(old.Type) { |
| 71 | continue |
| 72 | } |
| 73 | if p.evicted == nil { |
| 74 | p.evicted = map[PeerID]uint64{} |
| 75 | } |
| 76 | if old.Seq > p.evicted[old.From] { |
| 77 | p.evicted[old.From] = old.Seq |
| 78 | } |
| 79 | } |
| 80 | p.ring = p.ring[len(p.ring)-RingSize:] |
| 81 | } |
| 82 | |
| 83 | // Reset bumps the pair's generation and tells both sides. A member's |
| 84 | // standing is re-derived first, exactly as Relay does (§15.12). |
| 85 | func (h *Hub) Reset(ctx context.Context, sqid, session string, from, to PeerID) error { |
| 86 | gone, err := h.memberStandingGone(ctx, sqid, session, from, true) |
| 87 | if err != nil { |
| 88 | return err |
| 89 | } |
| 90 | if gone { |
| 91 | return ErrNotYours |
| 92 | } |
| 93 | h.mu.Lock() |
| 94 | defer h.mu.Unlock() |
| 95 | r, ok := h.rooms[sqid] |
| 96 | if !ok { |
| 97 | return ErrUnknownPeer |
| 98 | } |
| 99 | p, found := h.findLocked(sqid, session, from) |
| 100 | if !found { |
| 101 | return ErrUnknownPeer |
| 102 | } |
| 103 | if !p.Admitted { |
| 104 | return ErrNotAdmitted |
| 105 | } |
| 106 | if to == p.Peer { |
| 107 | return ErrUnknownPeer // a peer is not its own pair |
| 108 | } |
| 109 | tp, ok := r.peers[to] |
| 110 | if !ok || !tp.Admitted { |
| 111 | return ErrUnknownPeer |
| 112 | } |
| 113 | k := keyFor(p.Peer, tp.Peer) |
| 114 | h.genLocked(r, p.Peer, tp.Peer) // starts the pair at one if it is new |
| 115 | r.gens[k]++ |
| 116 | gen := r.gens[k] |
| 117 | for _, pair := range [][2]*Participation{{p, tp}, {tp, p}} { |
| 118 | m := Message{Type: TypePair, Peer: pair[1].Peer, Gen: gen, To: pair[0].Peer} |
| 119 | h.ringLocked(pair[0], m) |
| 120 | h.sendLocked(pair[0], m) |
| 121 | } |
| 122 | return nil |
| 123 | } |
| 124 | |
| 125 | // ResumeResult is what the stream handler owes a browser that came back |
| 126 | // inside its window: a fresh subscriber, and, when the participation is |
| 127 | // admitted, the roster, the replay and the senders it must reset with. |
| 128 | type ResumeResult struct { |
| 129 | Sub *Sub |
| 130 | Admitted bool |
| 131 | Roster *Roster |
| 132 | Replay []Message |
| 133 | Gap []PeerID |
| 134 | // ReportToken is the participation's, for the resumed hello (§15.26). |
| 135 | ReportToken string |
| 136 | } |
| 137 | |
| 138 | // ResumeError is a refused resume in the shape spec §4.5's table asks |
| 139 | // for: a status, a reason on the terminal 410s, and the boot id on the |
| 140 | // 404s so a browser can tell a restart from a lapsed window. |
| 141 | type ResumeError struct { |
| 142 | Status int |
| 143 | Reason Reason |
| 144 | Boot string |
| 145 | } |
| 146 | |
| 147 | func (e *ResumeError) Error() string { |
| 148 | return fmt.Sprintf("hub: resume refused %d %s", e.Status, e.Reason) |
| 149 | } |
| 150 | |
| 151 | // resumeRefusalLocked is spec §4.5's refusal ladder, and the only |
| 152 | // place it is written: Resume applies it before attaching, and |
| 153 | // CanResume applies it and stops there. |
| 154 | func (h *Hub) resumeRefusalLocked(sqid, session string, peer PeerID) *ResumeError { |
| 155 | if _, ok := h.rooms[sqid]; !ok { |
| 156 | return &ResumeError{Status: 404, Boot: h.boot} |
| 157 | } |
| 158 | // A tombstone is answered only through the room and session it was |
| 159 | // bound to. Why a participation ended is account state, and a peer |
| 160 | // id is published in every roster, so an ended peer's room-mate must |
| 161 | // not be able to read the reason off it (spec §4.2's binding rule). |
| 162 | if t, ok := h.tombstones[peer]; ok && t.room == sqid && t.session == session { |
| 163 | return &ResumeError{Status: 410, Reason: t.reason} |
| 164 | } |
| 165 | p, found := h.findLocked(sqid, session, peer) |
| 166 | if !found { |
| 167 | // Unknown here, unknown in another session, unknown in another |
| 168 | // room: one answer, so nothing leaks (spec §4.2). |
| 169 | return &ResumeError{Status: 404, Boot: h.boot} |
| 170 | } |
| 171 | // The window is thirty seconds, not thirty plus however long it is |
| 172 | // until the next tick: a resume that arrives after resume_until is |
| 173 | // refused now. Removing the participation and announcing the |
| 174 | // departure is still the tick's, so a refusal never has a room's |
| 175 | // events as a side effect. |
| 176 | if p.sub == nil && !p.resumeUntil.IsZero() && h.cfg.Now().After(p.resumeUntil) { |
| 177 | return &ResumeError{Status: 404, Boot: h.boot} |
| 178 | } |
| 179 | return nil |
| 180 | } |
| 181 | |
| 182 | // CanResume answers what a resume would answer, and does nothing else: |
| 183 | // it does not attach a subscriber, does not clear the resume window and |
| 184 | // does not read or trim the ring. It is the probe behind |
| 185 | // GET /m/{sqid}/stream?peer=&probe=1, which exists because EventSource |
| 186 | // hides the status code of a refused stream from the page. |
| 187 | // It re-derives a member's standing (§15.12) and answers 404 when it has |
| 188 | // gone — the same body an unknown peer gets — but it does not end the |
| 189 | // participation: that is the next mutating request's, or the tick's. A |
| 190 | // probe changes nothing, and a browser that reads 404 here goes on to |
| 191 | // open a fresh stream, which is what a dropped member should get. |
| 192 | func (h *Hub) CanResume(ctx context.Context, sqid, session string, peer PeerID) *ResumeError { |
| 193 | gone, err := h.memberStandingGone(ctx, sqid, session, peer, false) |
| 194 | if err != nil { |
| 195 | // A standing read that failed is not evidence of anything. The |
| 196 | // participation answers for itself, as it did before §15.12, and |
| 197 | // the tick will re-read standing in fifteen seconds. |
| 198 | h.cfg.Logger.Warn("hub: standing lookup failed on a resume probe", "room", sqid, "peer", peer, "error", err) |
| 199 | } |
| 200 | h.mu.Lock() |
| 201 | defer h.mu.Unlock() |
| 202 | if gone { |
| 203 | return &ResumeError{Status: 404, Boot: h.boot} |
| 204 | } |
| 205 | return h.resumeRefusalLocked(sqid, session, peer) |
| 206 | } |
| 207 | |
| 208 | // Resume reattaches a stream to a participation in its window. It is |
| 209 | // never a transition into the admitted state: a resume does not consume |
| 210 | // a cap slot and cannot admit anyone (spec §4.3). |
| 211 | // |
| 212 | // A member's standing is re-derived before anything is attached |
| 213 | // (§15.12). The refusal ladder runs first, so a participation that |
| 214 | // ended for a reason worth naming still answers 410 with that reason |
| 215 | // rather than spending a database read; a member whose standing has |
| 216 | // gone since is ended here and answered 404, which is the answer an |
| 217 | // unknown peer gets, byte for byte. |
| 218 | func (h *Hub) Resume(ctx context.Context, sqid, session string, peer PeerID, last map[PeerID]uint64) (ResumeResult, error) { |
| 219 | h.mu.Lock() |
| 220 | if re := h.resumeRefusalLocked(sqid, session, peer); re != nil { |
| 221 | h.mu.Unlock() |
| 222 | return ResumeResult{}, re |
| 223 | } |
| 224 | h.mu.Unlock() |
| 225 | |
| 226 | gone, err := h.memberStandingGone(ctx, sqid, session, peer, true) |
| 227 | if err != nil { |
| 228 | return ResumeResult{}, err |
| 229 | } |
| 230 | if gone { |
| 231 | return ResumeResult{}, &ResumeError{Status: 404, Boot: h.boot} |
| 232 | } |
| 233 | |
| 234 | h.mu.Lock() |
| 235 | defer h.mu.Unlock() |
| 236 | // The lock was down for the standing read, so the ladder is walked |
| 237 | // again: the window may have passed, or the meeting ended, while it |
| 238 | // was out. |
| 239 | if re := h.resumeRefusalLocked(sqid, session, peer); re != nil { |
| 240 | return ResumeResult{}, re |
| 241 | } |
| 242 | p, _ := h.findLocked(sqid, session, peer) |
| 243 | r := h.rooms[sqid] |
| 244 | if p.sub != nil { |
| 245 | // A second stream for the same peer replaces the attachment only: |
| 246 | // no tombstone, no departure, the participation carries on. |
| 247 | p.sub.terminate(Message{Type: TypeLost, Reason: ReasonReplaced}) |
| 248 | } |
| 249 | p.sub = newSub() |
| 250 | p.resumeUntil = time.Time{} |
| 251 | res := ResumeResult{Sub: p.sub, Admitted: p.Admitted, ReportToken: p.reportToken} |
| 252 | if !p.Admitted { |
| 253 | // A waiting guest learns nothing by resuming (ruling 9). |
| 254 | return res, nil |
| 255 | } |
| 256 | res.Roster = h.rosterFor(r, p) |
| 257 | // Replay is per sender: everything above last, unless the ring has |
| 258 | // dropped something above last, in which case that sender gets a gap |
| 259 | // and no partial replay (spec §4.5). |
| 260 | // |
| 261 | // Everything here is filtered against the roster that has just been |
| 262 | // built. The ring and the eviction marks outlive the peers that |
| 263 | // filled them, and the roster is the authority a browser reconciles |
| 264 | // against (§4.5): it has already closed its connection to a peer |
| 265 | // that is gone, so it must not be told to reset with it, replay its |
| 266 | // signalling, or rebuild its pair. |
| 267 | gapped := map[PeerID]bool{} |
| 268 | for from, high := range p.evicted { |
| 269 | if _, present := rosterGen(res.Roster, from); !present { |
| 270 | continue |
| 271 | } |
| 272 | if high > last[from] { |
| 273 | gapped[from] = true |
| 274 | res.Gap = append(res.Gap, from) |
| 275 | } |
| 276 | } |
| 277 | for _, m := range p.ring { |
| 278 | switch { |
| 279 | case isTargeted(m.Type): |
| 280 | if _, present := rosterGen(res.Roster, m.From); !present { |
| 281 | continue |
| 282 | } |
| 283 | if !gapped[m.From] && m.Seq > last[m.From] { |
| 284 | res.Replay = append(res.Replay, m) |
| 285 | } |
| 286 | case m.Type == TypePair: |
| 287 | // The roster already carries the current generation for every |
| 288 | // admitted peer, so this never fires for a current roster; |
| 289 | // the branch is kept because the ring is the authority for |
| 290 | // what was sent, and the browser reconciles from both (§4.6). |
| 291 | gen, present := rosterGen(res.Roster, m.Peer) |
| 292 | if present && m.Gen > gen { |
| 293 | res.Replay = append(res.Replay, m) |
| 294 | } |
| 295 | } |
| 296 | } |
| 297 | return res, nil |
| 298 | } |
| 299 | |
| 300 | // rosterGen is the recipient's generation with one peer, and whether |
| 301 | // that peer is in the roster at all. |
| 302 | func rosterGen(ro *Roster, peer PeerID) (uint64, bool) { |
| 303 | for _, rp := range ro.Peers { |
| 304 | if rp.Peer == peer { |
| 305 | return rp.Gen, true |
| 306 | } |
| 307 | } |
| 308 | return 0, false |
| 309 | } |
| 310 | |
amadan