| 1 | package rollout |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "sync" |
| 6 | "time" |
| 7 | ) |
| 8 | |
| 9 | // ackWriteBound bounds each config acknowledgement the armed scheduler |
| 10 | // writes (Task 11 rule 3, Astra R1 Flag triage T8): a store that hangs holds |
| 11 | // up the next ack this long, never the scheduler. Flag off, config sync's |
| 12 | // own writer (host.configAckFunc) is unchanged (plan review #11). |
| 13 | const ackWriteBound = 10 * time.Second |
| 14 | |
| 15 | // ackJob is one acknowledgement for the writer: seq is its place in the |
| 16 | // order of the config transitions that released it, taken under Manager.mu |
| 17 | // in the same critical section (Manager.ackLocked), and write is the store |
| 18 | // write, handed the bounded context. |
| 19 | type ackJob struct { |
| 20 | seq uint64 |
| 21 | write func(context.Context) |
| 22 | } |
| 23 | |
| 24 | // ackWriter is Manager.acks: the armed scheduler's acknowledgement writes, |
| 25 | // off the paths that release them. A config activation releases its held ack |
| 26 | // in done() before the attempt's slot frees, and in the scheduler loop for a |
| 27 | // sleeper (configSlept); neither may wait on the store, so they submit here |
| 28 | // and one drain goroutine writes, with no lock held while it writes (Task 11 |
| 29 | // rule 3). |
| 30 | // |
| 31 | // The order is fixed at the transition, not at the write (plan review #10): |
| 32 | // a host's acks only ever replace each other, so the writer keeps the newest |
| 33 | // pending one per host, by sequence, and drops an older one, whether it |
| 34 | // arrives after a newer one is pending or after a newer one was written. |
| 35 | // An ack written after a newer one would name a revision that is no longer |
| 36 | // the delivered one (spec §8.1). Hosts are written in the order their |
| 37 | // pending ack was first queued. Submission never blocks. |
| 38 | type ackWriter struct { |
| 39 | ctx context.Context |
| 40 | logf func(format string, args ...any) |
| 41 | kick chan struct{} |
| 42 | done chan struct{} // closed when the drain goroutine has returned |
| 43 | |
| 44 | mu sync.Mutex |
| 45 | pending map[string]ackJob |
| 46 | order []string // hosts with a pending ack, first queued first |
| 47 | newest map[string]uint64 // per host, the newest sequence accepted |
| 48 | closing bool // Close: drain what is pending, then return |
| 49 | stop bool // return after the write in progress |
| 50 | writing bool |
| 51 | // onSubmit, when set, runs at every accepted submission, with no lock |
| 52 | // held: a test's window into the moment an ack is released. |
| 53 | onSubmit func(host string) |
| 54 | } |
| 55 | |
| 56 | // newAckWriter starts the drain goroutine. It runs until ctx ends or close |
| 57 | // returns; a write in progress when ctx ends is finished (its own context is |
| 58 | // ctx's, so the store call returns as soon as it notices) and nothing more is |
| 59 | // written. |
| 60 | func newAckWriter(ctx context.Context, logf func(format string, args ...any)) *ackWriter { |
| 61 | w := &ackWriter{ctx: ctx, logf: logf, kick: make(chan struct{}, 1), done: make(chan struct{}), |
| 62 | pending: map[string]ackJob{}, newest: map[string]uint64{}} |
| 63 | go w.drain() |
| 64 | return w |
| 65 | } |
| 66 | |
| 67 | // submit queues j as host's ack unless an ack with a newer sequence was |
| 68 | // already queued or written for host. It never blocks. |
| 69 | func (w *ackWriter) submit(host string, j ackJob) { |
| 70 | w.mu.Lock() |
| 71 | if w.stop || j.seq <= w.newest[host] { |
| 72 | w.mu.Unlock() |
| 73 | return |
| 74 | } |
| 75 | w.newest[host] = j.seq |
| 76 | if _, queued := w.pending[host]; !queued { |
| 77 | w.order = append(w.order, host) |
| 78 | } |
| 79 | w.pending[host] = j |
| 80 | hook := w.onSubmit |
| 81 | w.mu.Unlock() |
| 82 | if hook != nil { |
| 83 | hook(host) |
| 84 | } |
| 85 | select { |
| 86 | case w.kick <- struct{}{}: |
| 87 | default: |
| 88 | } |
| 89 | } |
| 90 | |
| 91 | // next takes the first queued host's ack. ok is false when there is none, |
| 92 | // or the writer is to stop. |
| 93 | func (w *ackWriter) next() (host string, j ackJob, ok bool) { |
| 94 | w.mu.Lock() |
| 95 | defer w.mu.Unlock() |
| 96 | if w.stop || len(w.order) == 0 || w.ctx.Err() != nil { |
| 97 | w.writing = false |
| 98 | return "", ackJob{}, false |
| 99 | } |
| 100 | host, w.order = w.order[0], w.order[1:] |
| 101 | j = w.pending[host] |
| 102 | delete(w.pending, host) |
| 103 | w.writing = true |
| 104 | return host, j, true |
| 105 | } |
| 106 | |
| 107 | func (w *ackWriter) drain() { |
| 108 | defer close(w.done) |
| 109 | for { |
| 110 | for { |
| 111 | _, j, ok := w.next() |
| 112 | if !ok { |
| 113 | break |
| 114 | } |
| 115 | ctx, cancel := context.WithTimeout(w.ctx, ackWriteBound) |
| 116 | j.write(ctx) |
| 117 | cancel() |
| 118 | } |
| 119 | w.mu.Lock() |
| 120 | finished := w.stop || (w.closing && len(w.order) == 0) |
| 121 | w.mu.Unlock() |
| 122 | if finished { |
| 123 | return |
| 124 | } |
| 125 | select { |
| 126 | case <-w.ctx.Done(): |
| 127 | return |
| 128 | case <-w.kick: |
| 129 | } |
| 130 | } |
| 131 | } |
| 132 | |
| 133 | // idle reports whether nothing is pending or being written. |
| 134 | func (w *ackWriter) idle() bool { |
| 135 | w.mu.Lock() |
| 136 | defer w.mu.Unlock() |
| 137 | return len(w.order) == 0 && !w.writing |
| 138 | } |
| 139 | |
| 140 | // close writes what is pending and stops the writer, waiting until ctx |
| 141 | // ends. Once ctx ends the writer stops after the write in progress, and |
| 142 | // close returns without waiting for it: what is still pending then is not |
| 143 | // written. An ack submitted after close is dropped. |
| 144 | func (w *ackWriter) close(ctx context.Context) { |
| 145 | w.mu.Lock() |
| 146 | w.closing = true |
| 147 | w.mu.Unlock() |
| 148 | select { |
| 149 | case w.kick <- struct{}{}: |
| 150 | default: |
| 151 | } |
| 152 | select { |
| 153 | case <-w.done: |
| 154 | case <-ctx.Done(): |
| 155 | } |
| 156 | w.mu.Lock() |
| 157 | w.stop = true |
| 158 | left := len(w.order) |
| 159 | w.mu.Unlock() |
| 160 | if left > 0 { |
| 161 | w.logf("rollout: %d config acknowledgement(s) not written before the agent stopped", left) |
| 162 | } |
| 163 | } |
| 164 | |
amadan