| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "log" |
| 6 | "sync" |
| 7 | "sync/atomic" |
| 8 | "time" |
| 9 | |
| 10 | "github.com/carlosframework/platform/internal/activator" |
| 11 | "github.com/carlosframework/platform/internal/registry" |
| 12 | "github.com/carlosframework/platform/internal/release" |
| 13 | "github.com/carlosframework/platform/internal/rollout" |
| 14 | ) |
| 15 | |
| 16 | // relayGiveUp is how long the relay asks about one head. Root can fail to |
| 17 | // answer for good (its unit is broken, or the head is one it will never |
| 18 | // stage and never refuse), and without a bound the agent would touch the |
| 19 | // marker every stageRequestEvery for as long as the head stayed on the |
| 20 | // channel, each touch starting a root pass. After an hour it stops and says |
| 21 | // so; the next blocked pass that still finds the head unstaged notes it |
| 22 | // again, which starts a fresh hour (Astra R1 T15). |
| 23 | const relayGiveUp = time.Hour |
| 24 | |
| 25 | // stageRelay retransmits the stage request (spec §7.7). carlos-stage.path |
| 26 | // does not act on a change made while carlos-stage.service is running, so a |
| 27 | // touch that lands mid-pass is lost; this asks again every stageRequestEvery |
| 28 | // until root has recorded an outcome for the head: its stamp shows the head |
| 29 | // installed, or its refusal record names it. Armed only with the flag on. |
| 30 | type stageRelay struct { |
| 31 | mu sync.Mutex |
| 32 | pending map[string]pendingHead // host -> the channel head a pass could not stage |
| 33 | get func(host string) (*registry.Route, error) |
| 34 | // refused reads root's refusal record for a host (release.ReadRefusal |
| 35 | // over the data dir): nil when there is none. |
| 36 | refused func(host string) (*release.Refusal, error) |
| 37 | touch func() |
| 38 | // now and logf are the clock and the log line; nil is time.Now and |
| 39 | // log.Printf. They are seams for the give-up. |
| 40 | now func() time.Time |
| 41 | logf func(format string, args ...any) |
| 42 | } |
| 43 | |
| 44 | func (r *stageRelay) clock() time.Time { |
| 45 | if r.now != nil { |
| 46 | return r.now() |
| 47 | } |
| 48 | return time.Now() |
| 49 | } |
| 50 | |
| 51 | func (r *stageRelay) log(format string, args ...any) { |
| 52 | if r.logf != nil { |
| 53 | r.logf(format, args...) |
| 54 | return |
| 55 | } |
| 56 | log.Printf(format, args...) |
| 57 | } |
| 58 | |
| 59 | // pendingHead is a head the relay is asking root to stage, with the version |
| 60 | // the host had installed when it was noted (the baseline). Root records an |
| 61 | // outcome for a host by moving its installed Version, so a Version that is |
| 62 | // neither the head nor the baseline means root did something else, and a |
| 63 | // head still worth staging will be noted again by the next blocked pass. |
| 64 | type pendingHead struct { |
| 65 | ver string |
| 66 | base string |
| 67 | baseSet bool // false while the row could not be read at note time |
| 68 | // since is when the relay first noted this head: relayGiveUp counts |
| 69 | // from it. A head noted again keeps it; a newer head starts its own. |
| 70 | since time.Time |
| 71 | } |
| 72 | |
| 73 | // installed is the host's installed Version as of now: ok is false when the |
| 74 | // row cannot be read, and a vanished route reads as "". |
| 75 | func (r *stageRelay) installed(host string) (ver string, ok bool) { |
| 76 | rt, err := r.get(host) |
| 77 | if err != nil { |
| 78 | return "", false |
| 79 | } |
| 80 | if rt == nil { |
| 81 | return "", true |
| 82 | } |
| 83 | return rt.Version, true |
| 84 | } |
| 85 | |
| 86 | // note records the channel head a pass could not stage. A newer head for the |
| 87 | // same host replaces the older one: only the newest is worth asking for. The |
| 88 | // same head noted again only refreshes the baseline, and only when it has |
| 89 | // changed and the row can be read. |
| 90 | func (r *stageRelay) note(host, ver string) { |
| 91 | base, ok := r.installed(host) // outside the lock: a registry read |
| 92 | r.mu.Lock() |
| 93 | defer r.mu.Unlock() |
| 94 | p, have := r.pending[host] |
| 95 | if have && p.ver == ver { |
| 96 | if ok && (!p.baseSet || p.base != base) { |
| 97 | p.base, p.baseSet = base, true |
| 98 | r.pending[host] = p |
| 99 | } |
| 100 | return |
| 101 | } |
| 102 | r.pending[host] = pendingHead{ver: ver, base: base, baseSet: ok, since: r.clock()} |
| 103 | } |
| 104 | |
| 105 | // tick touches the marker once if any head is still outstanding. A head is |
| 106 | // dropped when root has installed it, when the host's installed Version has |
| 107 | // moved off the baseline (root recorded some outcome, for instance by |
| 108 | // installing a newer head straight from the channel), when root's refusal |
| 109 | // record names it, or when the route is gone. Without the baseline rule a |
| 110 | // head that was overtaken would be asked for every stageRequestEvery, each |
| 111 | // touch starting a root pass, until the route went. A row or a record that |
| 112 | // cannot be read is no outcome, so it keeps the head pending, until |
| 113 | // relayGiveUp has passed since it was first noted: then it is dropped with a |
| 114 | // log line, and a later blocked pass notes it afresh. |
| 115 | func (r *stageRelay) tick() bool { |
| 116 | r.mu.Lock() |
| 117 | snap := make(map[string]pendingHead, len(r.pending)) |
| 118 | for h, p := range r.pending { |
| 119 | snap[h] = p |
| 120 | } |
| 121 | r.mu.Unlock() |
| 122 | outstanding := false |
| 123 | for host, p := range snap { |
| 124 | if waited := r.clock().Sub(p.since); waited >= relayGiveUp { |
| 125 | r.mu.Lock() |
| 126 | cur, have := r.pending[host] |
| 127 | gone := have && cur.ver == p.ver && cur.since.Equal(p.since) // a newer head noted meanwhile stays |
| 128 | if gone { |
| 129 | delete(r.pending, host) |
| 130 | } |
| 131 | r.mu.Unlock() |
| 132 | if gone { |
| 133 | r.log("host: stage relay: giving up on %s for %s after %v: root has neither installed nor refused it; the next blocked pass will ask again", p.ver, host, waited.Round(time.Second)) |
| 134 | } |
| 135 | continue |
| 136 | } |
| 137 | rt, err := r.get(host) |
| 138 | if err != nil { |
| 139 | outstanding = true |
| 140 | continue |
| 141 | } |
| 142 | done := rt == nil || rt.Version == p.ver |
| 143 | if !done && !p.baseSet { |
| 144 | // The row was unreadable at note time: this read is the |
| 145 | // earliest baseline there is. |
| 146 | r.mu.Lock() |
| 147 | if cur, ok := r.pending[host]; ok && cur == p { |
| 148 | cur.base, cur.baseSet = rt.Version, true |
| 149 | r.pending[host] = cur |
| 150 | } |
| 151 | r.mu.Unlock() |
| 152 | } else if !done { |
| 153 | done = rt.Version != p.base |
| 154 | } |
| 155 | if !done { |
| 156 | rec, rerr := r.refused(host) |
| 157 | if rerr != nil { |
| 158 | outstanding = true |
| 159 | continue |
| 160 | } |
| 161 | done = rec != nil && rec.Version == p.ver |
| 162 | } |
| 163 | if done { |
| 164 | r.mu.Lock() |
| 165 | if r.pending[host].ver == p.ver { // a newer head noted meanwhile stays |
| 166 | delete(r.pending, host) |
| 167 | } |
| 168 | r.mu.Unlock() |
| 169 | continue |
| 170 | } |
| 171 | outstanding = true |
| 172 | } |
| 173 | if outstanding { |
| 174 | r.touch() |
| 175 | } |
| 176 | return outstanding |
| 177 | } |
| 178 | |
| 179 | // run ticks every interval until ctx ends. It is its own goroutine's body, so |
| 180 | // a slow registry read delays the next retransmission and nothing else. |
| 181 | func (r *stageRelay) run(ctx context.Context, every time.Duration) { |
| 182 | t := time.NewTicker(every) |
| 183 | defer t.Stop() |
| 184 | for { |
| 185 | select { |
| 186 | case <-ctx.Done(): |
| 187 | return |
| 188 | case <-t.C: |
| 189 | r.tick() |
| 190 | } |
| 191 | } |
| 192 | } |
| 193 | |
| 194 | // armedStageRelay is the relay stageRequester feeds; nil while the rollout |
| 195 | // is not armed, which leaves stageRequester exactly as it was. |
| 196 | var armedStageRelay atomic.Pointer[stageRelay] |
| 197 | |
| 198 | // newStageRelay is the relay over root's real records: the registry's stamp |
| 199 | // and the refusal files in the agent's data dir (its bake dir). |
| 200 | func newStageRelay(reg *registry.Registry, dd string) *stageRelay { |
| 201 | return &stageRelay{pending: map[string]pendingHead{}, get: reg.Get, |
| 202 | refused: func(host string) (*release.Refusal, error) { return release.ReadRefusal(dd, host) }, |
| 203 | touch: func() { touchStageMarker(dd) }} |
| 204 | } |
| 205 | |
| 206 | // armStageRelay starts the retransmission loop; it ends with ctx. |
| 207 | func armStageRelay(ctx context.Context, reg *registry.Registry, dd string) { |
| 208 | r := newStageRelay(reg, dd) |
| 209 | armedStageRelay.Store(r) |
| 210 | go r.run(ctx, stageRequestEvery) |
| 211 | } |
| 212 | |
| 213 | // armRolloutAndRelay arms the scheduler and, only when it armed, the stage |
| 214 | // relay (spec §7.7): a stage touch lost to a running root pass is asked |
| 215 | // again. With CARLOS_SWAP off armedStageRelay stays nil and the stage path is |
| 216 | // byte-for-byte what it was. |
| 217 | func armRolloutAndRelay(ctx context.Context, getenv func(string) string, svc *activator.Service, reg *registry.Registry, dd string, |
| 218 | sink activator.EventSink, openStore func(context.Context) (release.Store, error)) *rollout.Manager { |
| 219 | m := armRollout(ctx, getenv, svc, reg, dd, sink, openStore) |
| 220 | if m != nil { |
| 221 | armStageRelay(ctx, reg, dd) |
| 222 | } |
| 223 | return m |
| 224 | } |
| 225 | |
amadan