| 1 | package release |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "os" |
| 8 | "sort" |
| 9 | "sync" |
| 10 | "time" |
| 11 | ) |
| 12 | |
| 13 | // ChangeIndexBackstop is the longest a gated reader goes without a successful |
| 14 | // full read of a scope, however quiet the change index is (R3). It lives here |
| 15 | // so provision, host, edge and console share one value without import cycles. |
| 16 | // Callers jitter it by +-10%. |
| 17 | const ChangeIndexBackstop = 30 * time.Minute |
| 18 | |
| 19 | // ChangeIndexPollTimeout bounds one Poll from a caller with a deadline of its |
| 20 | // own to keep (the agent's fast tick, the edge's health-stamping loops): the |
| 21 | // wait for another caller's in-flight poll and the conditional GET together. |
| 22 | // A poll that runs out is "unknown" to its caller, never "unchanged". |
| 23 | const ChangeIndexPollTimeout = 5 * time.Second |
| 24 | |
| 25 | // ChangeWatch follows the change index (ChangeIndexKey) with one conditional |
| 26 | // GET per Poll: an unchanged index costs a 304 and no body. It is safe for |
| 27 | // concurrent use. |
| 28 | // |
| 29 | // The watch is unusable until its first good poll, and again after any |
| 30 | // failed one: an access error, a network error, an unknown schema, a |
| 31 | // malformed body, a store without ConditionalGetter, or a caller's context |
| 32 | // ending, while waiting for another caller's poll included. "Unknown" is never |
| 33 | // read as "empty and unchanged"; callers walk as they did before the index |
| 34 | // existed. The one exception is an absent index, which is the ordinary state |
| 35 | // before the first bump and reads as usable and empty. |
| 36 | type ChangeWatch struct { |
| 37 | store Store |
| 38 | |
| 39 | // pollSem serializes polls: a one-slot semaphore held across the |
| 40 | // network call. A channel rather than a mutex so a waiter behind a hung |
| 41 | // poll gives up when its context ends instead of inheriting the hang. |
| 42 | pollSem chan struct{} |
| 43 | |
| 44 | mu sync.Mutex // guards the fields below; never held across I/O |
| 45 | usable bool |
| 46 | etag string // etag of the last good index body ("" when it was absent) |
| 47 | entries map[string]uint64 // the last good observation |
| 48 | } |
| 49 | |
| 50 | // NewChangeWatch returns a watch over s. It has not polled yet, so it is |
| 51 | // unusable. |
| 52 | func NewChangeWatch(s Store) *ChangeWatch { |
| 53 | return &ChangeWatch{store: s, entries: map[string]uint64{}, pollSem: make(chan struct{}, 1)} |
| 54 | } |
| 55 | |
| 56 | // Poll reads the index conditionally and returns, sorted, the scopes whose |
| 57 | // generation differs from the last good observation: every scope on the |
| 58 | // first good poll, then those that rose or appeared. On a 304 it returns |
| 59 | // nil, nil. Any error marks the watch unusable until the next good poll and |
| 60 | // is returned; that next poll reports every scope that differs from the last |
| 61 | // good observation, including ones that rose while the watch was unusable. |
| 62 | // |
| 63 | // moved is advisory. Several consumers in one process share a watch (a box |
| 64 | // agent's tick, the record view, the cert mirror), so whichever polls first |
| 65 | // consumes a moved list and the others see nil. A consumer that must act on |
| 66 | // movement compares Observed against the map it last acted on instead of |
| 67 | // relying on Poll's return. |
| 68 | // |
| 69 | // A Poll that waits behind another caller's in-flight poll stops waiting when |
| 70 | // ctx ends and returns ctx's error. That give-up is a failed poll like any |
| 71 | // other: the watch is unusable until the next good poll, so no gated reader |
| 72 | // trusts an observation a stalled index GET has frozen. The ETag and entries |
| 73 | // are kept, so the next good poll of an unchanged index is a 304 again. |
| 74 | func (w *ChangeWatch) Poll(ctx context.Context) (moved []string, err error) { |
| 75 | if err := ctx.Err(); err != nil { |
| 76 | return nil, w.fail(err) |
| 77 | } |
| 78 | select { |
| 79 | case w.pollSem <- struct{}{}: |
| 80 | case <-ctx.Done(): |
| 81 | return nil, w.fail(ctx.Err()) |
| 82 | } |
| 83 | defer func() { <-w.pollSem }() |
| 84 | |
| 85 | g, ok := w.store.(ConditionalGetter) |
| 86 | if !ok { |
| 87 | return nil, w.fail(fmt.Errorf("release: change index unreadable: store has no conditional read: %w", errors.ErrUnsupported)) |
| 88 | } |
| 89 | w.mu.Lock() |
| 90 | etag := w.etag |
| 91 | w.mu.Unlock() |
| 92 | |
| 93 | body, newETag, changed, err := g.GetIfChanged(ctx, ChangeIndexKey, etag) |
| 94 | if errors.Is(err, os.ErrNotExist) { |
| 95 | return w.adopt(ChangeIndex{Schema: 1, Entries: map[string]uint64{}}, ""), nil |
| 96 | } |
| 97 | if err != nil { |
| 98 | return nil, w.fail(err) |
| 99 | } |
| 100 | if !changed { |
| 101 | w.mu.Lock() |
| 102 | w.usable = true // same bytes as the last good body, which are the last good observation |
| 103 | w.mu.Unlock() |
| 104 | return nil, nil |
| 105 | } |
| 106 | idx, _, err := readChangeIndex(ctx, func(context.Context, string) ([]byte, string, error) { |
| 107 | return body, newETag, nil |
| 108 | }) |
| 109 | if err != nil { |
| 110 | return nil, w.fail(err) |
| 111 | } |
| 112 | return w.adopt(idx, newETag), nil |
| 113 | } |
| 114 | |
| 115 | func (w *ChangeWatch) fail(err error) error { |
| 116 | w.mu.Lock() |
| 117 | w.usable = false |
| 118 | w.mu.Unlock() |
| 119 | return err |
| 120 | } |
| 121 | |
| 122 | // adopt records idx as the last good observation and returns the scopes it |
| 123 | // moved. |
| 124 | func (w *ChangeWatch) adopt(idx ChangeIndex, etag string) []string { |
| 125 | w.mu.Lock() |
| 126 | defer w.mu.Unlock() |
| 127 | var moved []string |
| 128 | for scope, gen := range idx.Entries { |
| 129 | if old, ok := w.entries[scope]; !ok || old != gen { |
| 130 | moved = append(moved, scope) |
| 131 | } |
| 132 | } |
| 133 | sort.Strings(moved) |
| 134 | w.entries = idx.Entries |
| 135 | w.etag = etag |
| 136 | w.usable = true |
| 137 | return moved |
| 138 | } |
| 139 | |
| 140 | // Usable reports whether the last poll succeeded. |
| 141 | func (w *ChangeWatch) Usable() bool { |
| 142 | w.mu.Lock() |
| 143 | defer w.mu.Unlock() |
| 144 | return w.usable |
| 145 | } |
| 146 | |
| 147 | // Gen is the last observed generation of scope. A scope the index does not |
| 148 | // list has generation 0 with ok true; ok is false while the watch is |
| 149 | // unusable. |
| 150 | func (w *ChangeWatch) Gen(scope string) (gen uint64, ok bool) { |
| 151 | w.mu.Lock() |
| 152 | defer w.mu.Unlock() |
| 153 | if !w.usable { |
| 154 | return 0, false |
| 155 | } |
| 156 | return w.entries[scope], true |
| 157 | } |
| 158 | |
| 159 | // Observed returns a copy of the last good entries, and false while the watch |
| 160 | // is unusable. Poll's moved list is advisory when consumers share a watch, so |
| 161 | // a consumer that must act on movement diffs Observed against its own |
| 162 | // last-acted map. |
| 163 | func (w *ChangeWatch) Observed() (map[string]uint64, bool) { |
| 164 | w.mu.Lock() |
| 165 | defer w.mu.Unlock() |
| 166 | if !w.usable { |
| 167 | return nil, false |
| 168 | } |
| 169 | out := make(map[string]uint64, len(w.entries)) |
| 170 | for k, v := range w.entries { |
| 171 | out[k] = v |
| 172 | } |
| 173 | return out, true |
| 174 | } |
| 175 | |
| 176 | // ReadGate decides, per scope, whether a full read is due: when the watch |
| 177 | // cannot vouch for the scope, the scope's generation has moved since the last |
| 178 | // successful read, or the last successful read is maxAge old. It is safe for |
| 179 | // concurrent use. |
| 180 | type ReadGate struct { |
| 181 | w *ChangeWatch |
| 182 | maxAge time.Duration |
| 183 | |
| 184 | mu sync.Mutex |
| 185 | marks map[string]gateMark |
| 186 | } |
| 187 | |
| 188 | type gateMark struct { |
| 189 | gen uint64 |
| 190 | at time.Time |
| 191 | ok bool |
| 192 | } |
| 193 | |
| 194 | // NewReadGate gates reads on w. A nil w makes every read due. maxAge is the |
| 195 | // longest a successful read stands in for the next one (use |
| 196 | // ChangeIndexBackstop, jittered). |
| 197 | func NewReadGate(w *ChangeWatch, maxAge time.Duration) *ReadGate { |
| 198 | return &ReadGate{w: w, maxAge: maxAge, marks: map[string]gateMark{}} |
| 199 | } |
| 200 | |
| 201 | // MaxAge is the longest a successful read stands in for the next one. A |
| 202 | // caller keeping finer marks of its own under one scope (the agent's canary |
| 203 | // groups, one per channel) ages them by it too. |
| 204 | func (g *ReadGate) MaxAge() time.Duration { return g.maxAge } |
| 205 | |
| 206 | // Due reports whether scope must be read now and returns the generation to |
| 207 | // pass to Mark once the read is done. Capture it BEFORE the read: a write that |
| 208 | // lands during the read then leaves the scope due on the next check instead |
| 209 | // of being hidden by it. The generation is 0 when the watch is nil or |
| 210 | // unusable. |
| 211 | func (g *ReadGate) Due(scope string, now time.Time) (gen uint64, due bool) { |
| 212 | if g == nil || g.w == nil { |
| 213 | return 0, true |
| 214 | } |
| 215 | gen, ok := g.w.Gen(scope) |
| 216 | if !ok { |
| 217 | return 0, true |
| 218 | } |
| 219 | g.mu.Lock() |
| 220 | m, marked := g.marks[scope] |
| 221 | g.mu.Unlock() |
| 222 | switch { |
| 223 | case !marked, !m.ok, m.gen != gen, now.Sub(m.at) >= g.maxAge: |
| 224 | return gen, true |
| 225 | } |
| 226 | return gen, false |
| 227 | } |
| 228 | |
| 229 | // Mark records the outcome of a read of scope that began when Due returned |
| 230 | // gen. A failed read (ok false) leaves the scope due. |
| 231 | func (g *ReadGate) Mark(scope string, gen uint64, now time.Time, ok bool) { |
| 232 | if g == nil { |
| 233 | return |
| 234 | } |
| 235 | g.mu.Lock() |
| 236 | g.marks[scope] = gateMark{gen: gen, at: now, ok: ok} |
| 237 | g.mu.Unlock() |
| 238 | } |
| 239 | |
amadan