| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "slices" |
| 7 | "sync" |
| 8 | "time" |
| 9 | |
| 10 | "github.com/carlosframework/platform/internal/controlsock" |
| 11 | "github.com/carlosframework/platform/internal/hostbox" |
| 12 | "github.com/carlosframework/platform/internal/hostclaim" |
| 13 | execlease "github.com/carlosframework/platform/internal/hostexec" |
| 14 | "github.com/carlosframework/platform/internal/hostwrite" |
| 15 | ) |
| 16 | |
| 17 | func executionExpectation(c *hostclaim.Claim) execlease.ExecutionRequest { |
| 18 | return execlease.ExecutionRequest{Schema: 1, Host: c.Host, Incarnation: c.Holder.Incarnation, Stamp: c.Holder.Stamp, Session: c.Holder.Session, Op: c.Op.ID} |
| 19 | } |
| 20 | |
| 21 | func (p *productionClaimLocal) executionTransport() func(context.Context, execlease.ExecutionRequest) (execlease.ExecutionResponse, error) { |
| 22 | if p.executionCall != nil { |
| 23 | return p.executionCall |
| 24 | } |
| 25 | return rootExecutionClient(controlsock.Path) |
| 26 | } |
| 27 | |
| 28 | func (p *productionClaimLocal) prepareWith(ctx context.Context, actor hostbox.Actor, o hostclaim.Observed) error { |
| 29 | if p.legacyParticipation { |
| 30 | return p.prepareLegacyWith(ctx, actor, o) |
| 31 | } |
| 32 | c := o.Claim |
| 33 | if c == nil { |
| 34 | return execlease.ErrState |
| 35 | } |
| 36 | p.mu.Lock() |
| 37 | old, lifetime, closed, assembly := p.slots[o.Host], p.lifetime, p.closed, p.assembly |
| 38 | delete(p.serving, o.Host) |
| 39 | p.mu.Unlock() |
| 40 | if closed || lifetime == nil || assembly == nil { |
| 41 | return hostwrite.ErrUnavailable |
| 42 | } |
| 43 | if p.Resources.Mode != "agent" { |
| 44 | return p.checkRootParticipation(ctx, actor, c) |
| 45 | } |
| 46 | if c.Free() || c.Op.Kind == hostclaim.KindServing || !slices.Contains(c.Op.Participants, actor.Box) { |
| 47 | if old != nil { |
| 48 | select { |
| 49 | case <-old.Ready: |
| 50 | case <-ctx.Done(): |
| 51 | return ctx.Err() |
| 52 | } |
| 53 | if old.Controller == nil { |
| 54 | return execlease.ErrLost |
| 55 | } |
| 56 | if err := old.Controller.Close(ctx); err != nil { |
| 57 | return err |
| 58 | } |
| 59 | p.mu.Lock() |
| 60 | if p.slots[o.Host] == old { |
| 61 | delete(p.slots, o.Host) |
| 62 | } |
| 63 | p.mu.Unlock() |
| 64 | } |
| 65 | if c.Free() || c.Op.Kind != hostclaim.KindServing { |
| 66 | return nil |
| 67 | } |
| 68 | q := executionExpectation(c) |
| 69 | q.Operation, q.ID = "executionView", "serving-view" |
| 70 | out, err := p.executionTransport()(ctx, q) |
| 71 | if err != nil { |
| 72 | return err |
| 73 | } |
| 74 | boot, err := execlease.BootID() |
| 75 | if err != nil { |
| 76 | return err |
| 77 | } |
| 78 | if out.Schema != 1 || out.Status != "serving" || out.Host != q.Host || out.Incarnation != q.Incarnation || out.Stamp != q.Stamp || out.Session != q.Session || out.Op != q.Op || out.BootID != boot || out.Token != "" || out.SafeUntil != 0 || out.LeaseETag != "" || !out.Expires.IsZero() { |
| 79 | return execlease.ErrLost |
| 80 | } |
| 81 | if err := execlease.ClearRootClientMirror(p.Resources.DataDir, q.Host); err != nil { |
| 82 | return err |
| 83 | } |
| 84 | p.mu.Lock() |
| 85 | defer p.mu.Unlock() |
| 86 | if p.closed || p.slots[o.Host] != nil { |
| 87 | return execlease.ErrLost |
| 88 | } |
| 89 | if p.serving == nil { |
| 90 | p.serving = make(map[string]execlease.ExecutionRequest) |
| 91 | } |
| 92 | p.serving[o.Host] = executionExpectation(c) |
| 93 | return nil |
| 94 | } |
| 95 | if c.Holder == nil { |
| 96 | return execlease.ErrState |
| 97 | } |
| 98 | if old != nil { |
| 99 | select { |
| 100 | case <-old.Ready: |
| 101 | case <-ctx.Done(): |
| 102 | return ctx.Err() |
| 103 | } |
| 104 | if old.Err != nil { |
| 105 | return old.Err |
| 106 | } |
| 107 | client, ok := old.Controller.(*execlease.RootClient) |
| 108 | if !ok { |
| 109 | return execlease.ErrLost |
| 110 | } |
| 111 | return client.Refresh(ctx) |
| 112 | } |
| 113 | boot, err := execlease.BootID() |
| 114 | if err != nil { |
| 115 | return err |
| 116 | } |
| 117 | slot := &claimParticipation{Ready: make(chan struct{}), Captured: o} |
| 118 | binding := execlease.Binding{Box: actor.Box, Enrollment: actor.Enrollment, Host: o.Host, Incarnation: c.Holder.Incarnation, Op: c.Op.ID, BootID: boot} |
| 119 | p.mu.Lock() |
| 120 | if p.closed || p.slots[o.Host] != nil { |
| 121 | p.mu.Unlock() |
| 122 | return execlease.ErrControllerBusy |
| 123 | } |
| 124 | if p.slots == nil { |
| 125 | p.slots = make(map[string]*claimParticipation) |
| 126 | } |
| 127 | p.slots[o.Host] = slot |
| 128 | p.mu.Unlock() |
| 129 | var captureMu sync.Mutex |
| 130 | captured := o |
| 131 | slot.Controller, slot.Err = execlease.OpenRootClient(lifetime, execlease.RootClientConfig{ |
| 132 | Binding: binding, Clock: execlease.SystemClock{}, Call: p.executionTransport(), |
| 133 | Project: func(s execlease.State) error { return execlease.ProjectRootClient(p.Resources.DataDir, s) }, |
| 134 | Observe: func(ctx context.Context) (execlease.ExecutionRequest, error) { |
| 135 | fresh, err := hostclaim.Read(ctx, p.Store, o.Host) |
| 136 | if err != nil { |
| 137 | return execlease.ExecutionRequest{}, err |
| 138 | } |
| 139 | if fresh.Absent || fresh.Claim == nil || fresh.Claim.Holder == nil { |
| 140 | return execlease.ExecutionRequest{}, execlease.ErrLost |
| 141 | } |
| 142 | if fresh.Claim.Holder.Incarnation == binding.Incarnation && fresh.Claim.Op.ID == binding.Op { |
| 143 | captureMu.Lock() |
| 144 | captured = fresh |
| 145 | captureMu.Unlock() |
| 146 | } |
| 147 | return executionExpectation(fresh.Claim), nil |
| 148 | }, |
| 149 | CloseAdmission: func() { p.cancelWork(o.Host) }, |
| 150 | Cleanup: func(ctx context.Context) error { |
| 151 | captureMu.Lock() |
| 152 | current := captured |
| 153 | captureMu.Unlock() |
| 154 | return p.cleanup(ctx, current, binding) |
| 155 | }, |
| 156 | }) |
| 157 | if slot.Err == nil { |
| 158 | var state execlease.State |
| 159 | state, slot.Err = slot.Controller.Ready(ctx) |
| 160 | slot.Binding = state.Binding |
| 161 | } |
| 162 | close(slot.Ready) |
| 163 | return slot.Err |
| 164 | } |
| 165 | |
| 166 | func (p *productionClaimLocal) Check(ctx context.Context, actor hostbox.Actor, c *hostclaim.Claim, selection hostbox.Selection) error { |
| 167 | if p.legacyParticipation { |
| 168 | return p.checkLegacy(ctx, actor, c, selection) |
| 169 | } |
| 170 | if ctx.Err() != nil { |
| 171 | return ctx.Err() |
| 172 | } |
| 173 | if c == nil || c.Holder == nil || actor.Box != selection.Box || actor.Region != selection.Region || c.Holder.Incarnation != selection.Captured.Incarnation || c.Holder.Stamp != selection.Captured.Stamp || slices.Contains(c.Lost, actor.Box) { |
| 174 | return hostbox.ErrChanged |
| 175 | } |
| 176 | p.mu.Lock() |
| 177 | slot, serving, closed := p.slots[c.Host], p.serving[c.Host], p.closed |
| 178 | p.mu.Unlock() |
| 179 | if closed { |
| 180 | return execlease.ErrLost |
| 181 | } |
| 182 | if p.Resources.Mode != "agent" { |
| 183 | return p.checkRootParticipation(ctx, actor, c) |
| 184 | } |
| 185 | if c.Op.Kind == hostclaim.KindServing { |
| 186 | if slot != nil || serving != executionExpectation(c) { |
| 187 | return execlease.ErrLost |
| 188 | } |
| 189 | return nil |
| 190 | } |
| 191 | if slot == nil { |
| 192 | return execlease.ErrInactive |
| 193 | } |
| 194 | select { |
| 195 | case <-slot.Ready: |
| 196 | default: |
| 197 | return execlease.ErrInactive |
| 198 | } |
| 199 | if slot.Err != nil { |
| 200 | return slot.Err |
| 201 | } |
| 202 | client, ok := slot.Controller.(*execlease.RootClient) |
| 203 | if !ok || slot.Binding.Box != actor.Box || slot.Binding.Enrollment != actor.Enrollment { |
| 204 | return execlease.ErrLost |
| 205 | } |
| 206 | return client.Check(ctx, executionExpectation(c)) |
| 207 | } |
| 208 | |
| 209 | // ackRefreshBefore is how close to its expiry an execution acknowledgement |
| 210 | // may get before this box refreshes it. The console completes a create, |
| 211 | // delete or purge only while every participant's acknowledgement is |
| 212 | // unexpired (hostops FirstPhase: now < Expires - RegistrationSkew), and its |
| 213 | // only retry is recovery, five minutes after the op started. The participant |
| 214 | // loop runs every 30 s and the lease (execlease.TTL, 90 s) renews every 20 s, |
| 215 | // so a just-written acknowledgement has at least 70 s left. Two-thirds of the |
| 216 | // TTL (60 s) leaves the first pass after a write alone and refreshes on the |
| 217 | // next one. A refresh is due at most 60 s before expiry and the next pass |
| 218 | // comes 30 s later, so the acknowledgement stays usable as long as a pass |
| 219 | // starts within 60 - 30 - 5 = 25 s of its tick. |
| 220 | const ackRefreshBefore = 2 * execlease.TTL / 3 |
| 221 | |
| 222 | func (p *productionClaimLocal) wallNow() time.Time { |
| 223 | if p.now != nil { |
| 224 | return p.now() |
| 225 | } |
| 226 | return time.Now() |
| 227 | } |
| 228 | |
| 229 | func (p *productionClaimLocal) acknowledgeOperation(ctx context.Context, actor hostbox.Actor, o hostclaim.Observed) error { |
| 230 | c := o.Claim |
| 231 | if c == nil || c.Holder == nil || (c.Op.Kind != hostclaim.KindCreating && c.Op.Kind != hostclaim.KindDeleting && c.Op.Kind != hostclaim.KindPurging) || !slices.Contains(c.Op.Participants, actor.Box) { |
| 232 | return nil |
| 233 | } |
| 234 | if previous := c.Op.Journal["exec_ack."+actor.Box]; previous != "" { |
| 235 | return p.refreshAcknowledgement(ctx, actor, o, previous) |
| 236 | } |
| 237 | p.mu.Lock() |
| 238 | slot := p.slots[o.Host] |
| 239 | p.mu.Unlock() |
| 240 | if slot == nil { |
| 241 | return execlease.ErrInactive |
| 242 | } |
| 243 | select { |
| 244 | case <-slot.Ready: |
| 245 | case <-ctx.Done(): |
| 246 | return ctx.Err() |
| 247 | } |
| 248 | if slot.Err != nil { |
| 249 | return slot.Err |
| 250 | } |
| 251 | if slot.Controller == nil { |
| 252 | return execlease.ErrInactive |
| 253 | } |
| 254 | state, err := slot.Controller.Ready(ctx) |
| 255 | if err != nil { |
| 256 | return err |
| 257 | } |
| 258 | if state.State != "active" || state.Host != o.Host || state.Incarnation != c.Holder.Incarnation || state.Op != c.Op.ID || state.Box != actor.Box || state.Enrollment != actor.Enrollment { |
| 259 | return execlease.ErrLost |
| 260 | } |
| 261 | ack := hostclaim.ExecutionAcknowledgement{Op: state.Op, Enrollment: state.Enrollment, BootID: state.BootID, ControllerID: state.ControllerID, LeaseETag: state.LeaseETag, Expires: state.Expires} |
| 262 | // Recorded before the write: an uncertain write that landed is still this |
| 263 | // controller's acknowledgement on the next pass. |
| 264 | p.mu.Lock() |
| 265 | slot.AckController = state.ControllerID |
| 266 | p.mu.Unlock() |
| 267 | _, err = hostclaim.Update(ctx, p.Store, o, hostclaim.AcknowledgeExecution(actor.Box, ack)) |
| 268 | return err |
| 269 | } |
| 270 | |
| 271 | // refreshAcknowledgement extends this controller's own acknowledgement for |
| 272 | // the op while it waits on the console, once fewer than ackRefreshBefore |
| 273 | // remain. It writes only from the controller's current Ready state, under the |
| 274 | // same checks as the first acknowledgement, and never on behalf of another |
| 275 | // boot or controller: an acknowledgement this slot did not write is left to |
| 276 | // expire. |
| 277 | // |
| 278 | // The root broker rotates the controller token on every renewal, so the |
| 279 | // current state's ControllerID changes every 20 s. The slot is what |
| 280 | // identifies the same controller; the refreshed acknowledgement keeps the |
| 281 | // ControllerID the first one bound and carries the current lease ETag and |
| 282 | // expiry. |
| 283 | func (p *productionClaimLocal) refreshAcknowledgement(ctx context.Context, actor hostbox.Actor, o hostclaim.Observed, previous string) error { |
| 284 | c := o.Claim |
| 285 | var prior hostclaim.ExecutionAcknowledgement |
| 286 | if json.Unmarshal([]byte(previous), &prior) != nil || prior.Op != c.Op.ID || prior.Enrollment != actor.Enrollment || prior.BootID == "" || prior.ControllerID == "" { |
| 287 | return nil |
| 288 | } |
| 289 | if !p.wallNow().Add(ackRefreshBefore).After(prior.Expires) { |
| 290 | return nil |
| 291 | } |
| 292 | p.mu.Lock() |
| 293 | slot := p.slots[o.Host] |
| 294 | p.mu.Unlock() |
| 295 | if slot == nil { |
| 296 | return nil |
| 297 | } |
| 298 | select { |
| 299 | case <-slot.Ready: |
| 300 | case <-ctx.Done(): |
| 301 | return ctx.Err() |
| 302 | } |
| 303 | p.mu.Lock() |
| 304 | owned := slot.AckController |
| 305 | p.mu.Unlock() |
| 306 | if slot.Err != nil || slot.Controller == nil || owned == "" || owned != prior.ControllerID { |
| 307 | return nil |
| 308 | } |
| 309 | state, err := slot.Controller.Ready(ctx) |
| 310 | if err != nil { |
| 311 | return err |
| 312 | } |
| 313 | if state.State != "active" || state.Host != o.Host || state.Incarnation != c.Holder.Incarnation || state.Op != c.Op.ID || state.Box != actor.Box || state.Enrollment != actor.Enrollment { |
| 314 | return execlease.ErrLost |
| 315 | } |
| 316 | if state.BootID != prior.BootID || state.LeaseETag == "" || !state.Expires.After(prior.Expires) { |
| 317 | return nil |
| 318 | } |
| 319 | ack := hostclaim.ExecutionAcknowledgement{Op: prior.Op, Enrollment: prior.Enrollment, BootID: prior.BootID, ControllerID: prior.ControllerID, LeaseETag: state.LeaseETag, Expires: state.Expires} |
| 320 | _, err = hostclaim.Update(ctx, p.Store, o, hostclaim.AcknowledgeExecution(actor.Box, ack)) |
| 321 | return err |
| 322 | } |
| 323 | |
amadan