| 1 | package transcript |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/rand" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "time" |
| 9 | |
| 10 | "reasonix/internal/eventwire" |
| 11 | ) |
| 12 | |
| 13 | const FollowProtocolVersion = 2 |
| 14 | |
| 15 | type Change struct { |
| 16 | Runtime *Runtime `json:"runtime,omitempty"` |
| 17 | AttemptID string `json:"attemptId,omitempty"` |
| 18 | Index uint64 `json:"index"` |
| 19 | ResultSeq uint64 `json:"resultSeq,omitempty"` |
| 20 | ResultKind string `json:"resultKind,omitempty"` |
| 21 | Revision uint64 `json:"revision"` |
| 22 | CommitSeq uint64 `json:"commitSeq"` |
| 23 | DurableSeq uint64 `json:"durableSeq"` |
| 24 | FirstSeq uint64 `json:"firstSeq,omitempty"` |
| 25 | Records []Message `json:"records,omitempty"` |
| 26 | Event *eventwire.Event `json:"event,omitempty"` |
| 27 | ResetRequired bool `json:"resetRequired,omitempty"` |
| 28 | } |
| 29 | |
| 30 | type FollowRequest struct { |
| 31 | Subscription string `json:"subscription,omitempty"` |
| 32 | AfterRevision uint64 `json:"afterRevision,omitempty"` |
| 33 | Close bool `json:"close,omitempty"` |
| 34 | } |
| 35 | |
| 36 | type FollowResponse struct { |
| 37 | ProtocolVersion int `json:"protocolVersion"` |
| 38 | Subscription string `json:"subscription"` |
| 39 | Snapshot *Snapshot `json:"snapshot,omitempty"` |
| 40 | Changes []Change `json:"changes"` |
| 41 | ResetRequired bool `json:"resetRequired"` |
| 42 | } |
| 43 | |
| 44 | type follower struct { |
| 45 | ack uint64 |
| 46 | queue []Change |
| 47 | bytes int |
| 48 | reset bool |
| 49 | wake chan struct{} |
| 50 | seen time.Time |
| 51 | } |
| 52 | |
| 53 | // Follow registers before freezing its initial cut. Subsequent calls long-poll |
| 54 | // this subscription; no network callback is invoked under the owner lock. |
| 55 | func (p *Projection) Follow(ctx context.Context, req FollowRequest) (FollowResponse, error) { |
| 56 | p.mu.Lock() |
| 57 | if p.followers == nil { |
| 58 | p.followers = make(map[string]*follower) |
| 59 | } |
| 60 | now := time.Now() |
| 61 | for id, f := range p.followers { |
| 62 | if now.Sub(f.seen) > 2*time.Minute { |
| 63 | f.reset = true |
| 64 | select { |
| 65 | case f.wake <- struct{}{}: |
| 66 | default: |
| 67 | } |
| 68 | delete(p.followers, id) |
| 69 | } |
| 70 | } |
| 71 | out := FollowResponse{ProtocolVersion: FollowProtocolVersion, Subscription: req.Subscription, Changes: []Change{}} |
| 72 | if req.Close { |
| 73 | if f := p.followers[req.Subscription]; f != nil { |
| 74 | f.reset = true |
| 75 | select { |
| 76 | case f.wake <- struct{}{}: |
| 77 | default: |
| 78 | } |
| 79 | } |
| 80 | delete(p.followers, req.Subscription) |
| 81 | p.mu.Unlock() |
| 82 | return out, nil |
| 83 | } |
| 84 | if req.Subscription == "" { |
| 85 | if len(p.followers) >= 64 { |
| 86 | p.mu.Unlock() |
| 87 | return out, errors.New("too many transcript followers") |
| 88 | } |
| 89 | out.Subscription = rand.Text() |
| 90 | f := &follower{wake: make(chan struct{}, 1), seen: now, ack: p.revision} |
| 91 | p.followers[out.Subscription] = f |
| 92 | frozen, err := p.freezeLocked("") |
| 93 | if err != nil { |
| 94 | delete(p.followers, out.Subscription) |
| 95 | } |
| 96 | p.mu.Unlock() |
| 97 | if err != nil { |
| 98 | return out, err |
| 99 | } |
| 100 | snapshot, err := frozen.snapshotCurrent(PageRequest{Records: 32}) |
| 101 | snapshot.ProtocolVersion = FollowProtocolVersion |
| 102 | out.Snapshot = &snapshot |
| 103 | return out, err |
| 104 | } |
| 105 | f := p.followers[req.Subscription] |
| 106 | if f == nil { |
| 107 | p.mu.Unlock() |
| 108 | out.ResetRequired = true |
| 109 | return out, nil |
| 110 | } |
| 111 | f.seen = now |
| 112 | if req.AfterRevision < f.ack || req.AfterRevision > p.revision { |
| 113 | f.reset = true |
| 114 | } else { |
| 115 | f.ack = req.AfterRevision |
| 116 | } |
| 117 | p.mu.Unlock() |
| 118 | timer := time.NewTimer(25 * time.Second) |
| 119 | defer timer.Stop() |
| 120 | for { |
| 121 | p.mu.Lock() |
| 122 | for len(f.queue) > 0 && f.queue[0].Revision <= req.AfterRevision { |
| 123 | encoded, _ := json.Marshal(f.queue[0]) |
| 124 | f.bytes -= len(encoded) |
| 125 | f.queue = f.queue[1:] |
| 126 | } |
| 127 | if f.reset || len(f.queue) > 0 { |
| 128 | out.ResetRequired = f.reset |
| 129 | out.Changes = append(out.Changes, f.queue...) |
| 130 | p.mu.Unlock() |
| 131 | return out, nil |
| 132 | } |
| 133 | p.mu.Unlock() |
| 134 | select { |
| 135 | case <-ctx.Done(): |
| 136 | return out, ctx.Err() |
| 137 | case <-timer.C: |
| 138 | return out, nil |
| 139 | case <-f.wake: |
| 140 | } |
| 141 | } |
| 142 | } |
| 143 | |
| 144 | func (p *Projection) CloseFollowers() { |
| 145 | p.mu.Lock() |
| 146 | defer p.mu.Unlock() |
| 147 | for _, f := range p.followers { |
| 148 | f.reset = true |
| 149 | select { |
| 150 | case f.wake <- struct{}{}: |
| 151 | default: |
| 152 | } |
| 153 | } |
| 154 | clear(p.followers) |
| 155 | } |
| 156 | |
| 157 | func (p *Projection) publishChangeLocked(change Change) { |
| 158 | change.Revision, change.CommitSeq, change.DurableSeq = p.revision, p.covered, p.durable |
| 159 | encoded, err := json.Marshal(change) |
| 160 | // Detach retained deltas from mutable event payloads and caller-owned rows. |
| 161 | var owned Change |
| 162 | if err == nil { |
| 163 | err = json.Unmarshal(encoded, &owned) |
| 164 | } |
| 165 | for _, f := range p.followers { |
| 166 | if f.reset { |
| 167 | continue |
| 168 | } |
| 169 | if err != nil || len(f.queue) >= 256 || f.bytes+len(encoded) > MaxResponseBytes/2 { |
| 170 | f.queue, f.bytes, f.reset = nil, 0, true |
| 171 | } else { |
| 172 | f.queue = append(f.queue, owned) |
| 173 | f.bytes += len(encoded) |
| 174 | } |
| 175 | select { |
| 176 | case f.wake <- struct{}{}: |
| 177 | default: |
| 178 | } |
| 179 | } |
| 180 | } |
| 181 | |
| 182 | func (p *Projection) SetDurableSequence(sequence uint64) { |
| 183 | p.mu.Lock() |
| 184 | defer p.mu.Unlock() |
| 185 | if sequence > p.durable { |
| 186 | p.durable = sequence |
| 187 | p.revision++ |
| 188 | p.publishChangeLocked(Change{}) |
| 189 | } |
| 190 | } |
| 191 |