| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "os/exec" |
| 7 | "time" |
| 8 | |
| 9 | "github.com/carlosframework/platform/internal/hostbox" |
| 10 | "github.com/carlosframework/platform/internal/hostclaim" |
| 11 | "github.com/carlosframework/platform/internal/hostwrite" |
| 12 | "github.com/carlosframework/platform/internal/registry" |
| 13 | "github.com/carlosframework/platform/internal/release" |
| 14 | ) |
| 15 | |
| 16 | type ServingRetries struct { |
| 17 | BeforeReserve func(context.Context, registry.Route, hostclaim.Observed) error |
| 18 | Claims release.Store |
| 19 | Box, Enrollment, BootID, ControllerID string |
| 20 | Now func() time.Time |
| 21 | NewID func() string |
| 22 | Inputs func(context.Context, registry.Route) (string, error) |
| 23 | StopAttempt func(context.Context, hostclaim.Observed, hostclaim.WakeRetryAttempt) error |
| 24 | ExactHealth func(context.Context, *exec.Cmd, hostclaim.WakeRetryAttempt) error |
| 25 | } |
| 26 | |
| 27 | func (d ServingRetries) valid() bool { |
| 28 | return d.Claims != nil && d.Box != "" && d.Enrollment != "" && d.BootID != "" && d.ControllerID != "" && d.Now != nil && d.NewID != nil && d.Inputs != nil && d.StopAttempt != nil && d.ExactHealth != nil |
| 29 | } |
| 30 | |
| 31 | func (d ServingRetries) attempt(inputs string, session uint64) hostclaim.WakeRetryAttempt { |
| 32 | return hostclaim.WakeRetryAttempt{ID: d.NewID(), Box: d.Box, Enrollment: d.Enrollment, BootID: d.BootID, ControllerID: d.ControllerID, Session: session, InvocationID: d.NewID(), LaunchInputs: inputs} |
| 33 | } |
| 34 | |
| 35 | func retryOf(o hostclaim.Observed) *hostclaim.WakeRetry { |
| 36 | if o.Claim == nil || o.Claim.Holder == nil { |
| 37 | return nil |
| 38 | } |
| 39 | return o.Claim.Holder.WakeRetry |
| 40 | } |
| 41 | |
| 42 | func (d ServingRetries) settle(ctx context.Context, expected hostclaim.Observed, writes ...func(hostclaim.Observed, hostclaim.Observed) error) error { |
| 43 | r := retryOf(expected) |
| 44 | if r == nil || r.State != "attempting" || r.Attempt == nil { |
| 45 | return nil |
| 46 | } |
| 47 | o, err := hostclaim.Read(ctx, d.Claims, expected.Host) |
| 48 | if err != nil { |
| 49 | return err |
| 50 | } |
| 51 | current := retryOf(o) |
| 52 | if o.Claim == nil || o.Claim.Holder == nil || o.Claim.Holder.Incarnation != expected.Claim.Holder.Incarnation || current == nil || current.EpisodeID != r.EpisodeID || current.Attempt == nil || *current.Attempt != *r.Attempt { |
| 53 | return hostbox.ErrChanged |
| 54 | } |
| 55 | if err := d.StopAttempt(ctx, o, *r.Attempt); err != nil { |
| 56 | return err |
| 57 | } |
| 58 | // A stop can change bookkeeping. Never use its earlier ETag for settlement. |
| 59 | o, err = hostclaim.Read(ctx, d.Claims, expected.Host) |
| 60 | if err != nil { |
| 61 | return err |
| 62 | } |
| 63 | written, err := hostclaim.Update(ctx, d.Claims, o, hostclaim.FailWakeRetryAttempt(hostclaim.WakeRetryResult{HolderIncarnation: expected.Claim.Holder.Incarnation, EpisodeID: r.EpisodeID, Attempt: *r.Attempt, Now: d.Now()})) |
| 64 | if err == nil { |
| 65 | for _, write := range writes { |
| 66 | if write != nil { |
| 67 | if err := write(o, written); err != nil { |
| 68 | return err |
| 69 | } |
| 70 | } |
| 71 | } |
| 72 | } |
| 73 | return err |
| 74 | } |
| 75 | |
| 76 | func (d ServingRetries) Wrap(base *ClaimRuntime) (*ClaimRuntime, error) { |
| 77 | if !d.valid() || base == nil || base.Admit == nil || base.Prepare == nil || base.Prerequisites == nil || base.FailedAdmission == nil { |
| 78 | return nil, hostwrite.ErrUnavailable |
| 79 | } |
| 80 | out := *base |
| 81 | out.Prerequisites = func(ctx context.Context, rt registry.Route, o hostclaim.Observed, p StartPolicy) error { |
| 82 | if o.Claim != nil && o.Claim.Op.Kind == hostclaim.KindServing && !p.Probe { |
| 83 | if r := retryOf(o); r != nil { |
| 84 | if r.State == "attempting" { |
| 85 | if err := d.settle(ctx, o); err != nil { |
| 86 | return err |
| 87 | } |
| 88 | return hostbox.ErrChanged |
| 89 | } |
| 90 | if r.State != "open" || d.Now().Before(r.NotBefore) { |
| 91 | return hostclaim.ErrWrongState |
| 92 | } |
| 93 | } |
| 94 | } |
| 95 | if o.Claim != nil && o.Claim.Op.Kind == hostclaim.KindServing && !p.Probe && d.BeforeReserve != nil { |
| 96 | if err := d.BeforeReserve(ctx, rt, o); err != nil { |
| 97 | return err |
| 98 | } |
| 99 | } |
| 100 | return base.Prerequisites(ctx, rt, o, p) |
| 101 | } |
| 102 | out.Admit = func(ctx context.Context, rt registry.Route, o hostclaim.Observed, p StartPolicy) (result hostclaim.Observed, resultErr error) { |
| 103 | if o.Claim == nil || o.Claim.Op.Kind != hostclaim.KindServing || p.Probe { |
| 104 | return base.Admit(ctx, rt, o, p) |
| 105 | } |
| 106 | cleanupObservation := o |
| 107 | defer func() { |
| 108 | if resultErr != nil { |
| 109 | recovery, cancel := CleanupContext(ctx, 10*time.Second) |
| 110 | defer cancel() |
| 111 | resultErr = errors.Join(resultErr, base.FailedAdmission(recovery, cleanupObservation, resultErr)) |
| 112 | } |
| 113 | }() |
| 114 | if d.BeforeReserve != nil { |
| 115 | if err := d.BeforeReserve(ctx, rt, o); err != nil { |
| 116 | return hostclaim.Observed{}, err |
| 117 | } |
| 118 | } |
| 119 | inputs, err := d.Inputs(ctx, rt) |
| 120 | if err != nil || inputs == "" { |
| 121 | return hostclaim.Observed{}, errors.Join(err, hostwrite.ErrUnavailable) |
| 122 | } |
| 123 | episode := d.NewID() |
| 124 | if r := retryOf(o); r != nil { |
| 125 | episode = r.EpisodeID |
| 126 | } |
| 127 | attempt := d.attempt(inputs, 0) |
| 128 | if d.BeforeReserve != nil { |
| 129 | if err := d.BeforeReserve(ctx, rt, o); err != nil { |
| 130 | return hostclaim.Observed{}, err |
| 131 | } |
| 132 | } |
| 133 | reserved, err := hostclaim.Update(ctx, d.Claims, o, hostclaim.ReserveWakeRetryAttempt(hostclaim.WakeRetryReservation{HolderIncarnation: o.Claim.Holder.Incarnation, EpisodeID: episode, Attempt: attempt, Now: d.Now()})) |
| 134 | if err != nil { |
| 135 | return hostclaim.Observed{}, err |
| 136 | } |
| 137 | cleanupObservation = reserved |
| 138 | admitted, err := base.Admit(ctx, rt, reserved, p) |
| 139 | if err == nil && admitted.Claim != nil && admitted.Claim.Holder != nil { |
| 140 | admitted, err = hostclaim.Update(ctx, d.Claims, admitted, hostclaim.BindWakeRetrySession(o.Claim.Holder.Incarnation, episode, attempt.ID, admitted.Claim.Holder.Session)) |
| 141 | } else if err == nil { |
| 142 | err = hostbox.ErrChanged |
| 143 | } |
| 144 | if err != nil { |
| 145 | recovery, cancel := CleanupContext(ctx, 10*time.Second) |
| 146 | defer cancel() |
| 147 | // An ambiguous admission or bind is settled using the stored attempt, never refunded. |
| 148 | fresh, readErr := hostclaim.Read(recovery, d.Claims, o.Host) |
| 149 | if readErr == nil && fresh.Claim != nil && fresh.Claim.Holder != nil && fresh.Claim.Holder.Incarnation == o.Claim.Holder.Incarnation { |
| 150 | r := retryOf(fresh) |
| 151 | if r != nil && r.EpisodeID == episode && r.Attempt != nil && r.Attempt.ID == attempt.ID { |
| 152 | readErr = d.settle(recovery, fresh) |
| 153 | } else { |
| 154 | readErr = hostbox.ErrChanged |
| 155 | } |
| 156 | } |
| 157 | return hostclaim.Observed{}, errors.Join(err, readErr) |
| 158 | } |
| 159 | return admitted, nil |
| 160 | } |
| 161 | out.FailedAdmission = func(ctx context.Context, o hostclaim.Observed, cause error) error { |
| 162 | err := base.FailedAdmission(ctx, o, cause) |
| 163 | fresh, readErr := hostclaim.Read(ctx, d.Claims, o.Host) |
| 164 | if readErr != nil { |
| 165 | return errors.Join(err, readErr) |
| 166 | } |
| 167 | if o.Claim == nil || o.Claim.Holder == nil || fresh.Claim == nil || fresh.Claim.Holder == nil || o.Claim.Holder.Incarnation != fresh.Claim.Holder.Incarnation { |
| 168 | return errors.Join(err, hostbox.ErrChanged) |
| 169 | } |
| 170 | expected, current := retryOf(o), retryOf(fresh) |
| 171 | if expected == nil || current == nil || expected.EpisodeID != current.EpisodeID { |
| 172 | return err |
| 173 | } |
| 174 | if current.Attempt != nil && (!sameAdmission(fresh, o) || current.Attempt.Box != d.Box || current.Attempt.Enrollment != d.Enrollment || current.Attempt.BootID != d.BootID || current.Attempt.ControllerID != d.ControllerID) { |
| 175 | return errors.Join(err, hostbox.ErrChanged) |
| 176 | } |
| 177 | return errors.Join(err, d.settle(ctx, fresh)) |
| 178 | } |
| 179 | out.Prepare = func(ctx context.Context, rt registry.Route, o hostclaim.Observed, p StartPolicy) (ClaimStart, error) { |
| 180 | s, err := base.Prepare(ctx, rt, o, p) |
| 181 | if err != nil || p.Probe || o.Claim == nil || o.Claim.Op.Kind != hostclaim.KindServing { |
| 182 | return s, err |
| 183 | } |
| 184 | r := retryOf(o) |
| 185 | if r == nil || r.Attempt == nil || r.Attempt.Session == 0 || r.Attempt.Session != o.Claim.Holder.Session { |
| 186 | return ClaimStart{}, hostbox.ErrChanged |
| 187 | } |
| 188 | episode, inc := r.EpisodeID, o.Claim.Holder.Incarnation |
| 189 | attempt := *r.Attempt |
| 190 | s.CurrentAttempt = func() hostclaim.WakeRetryAttempt { return attempt } |
| 191 | s.Reserve = func(ctx context.Context, n int) error { |
| 192 | if err := ctx.Err(); err != nil { |
| 193 | return err |
| 194 | } |
| 195 | fresh, err := hostclaim.Read(ctx, d.Claims, o.Host) |
| 196 | if err != nil { |
| 197 | return err |
| 198 | } |
| 199 | if !sameAdmission(fresh, o) { |
| 200 | return hostbox.ErrChanged |
| 201 | } |
| 202 | if n == 0 { |
| 203 | rr := retryOf(fresh) |
| 204 | if rr == nil || rr.EpisodeID != episode || rr.Attempt == nil || *rr.Attempt != attempt { |
| 205 | return hostbox.ErrChanged |
| 206 | } |
| 207 | return nil |
| 208 | } |
| 209 | inputs, err := d.Inputs(ctx, rt) |
| 210 | if err != nil { |
| 211 | return err |
| 212 | } |
| 213 | if inputs != attempt.LaunchInputs { |
| 214 | return hostbox.ErrChanged |
| 215 | } |
| 216 | next := d.attempt(inputs, o.Claim.Holder.Session) |
| 217 | written, err := hostclaim.Update(ctx, d.Claims, fresh, hostclaim.ReserveWakeRetryAttempt(hostclaim.WakeRetryReservation{HolderIncarnation: inc, EpisodeID: episode, Attempt: next, Now: d.Now()})) |
| 218 | if err == nil { |
| 219 | attempt = next |
| 220 | if s.RetryClaimWrite != nil { |
| 221 | return s.RetryClaimWrite(fresh, written) |
| 222 | } |
| 223 | } |
| 224 | return err |
| 225 | } |
| 226 | s.Failed = func(ctx context.Context, _ int) error { |
| 227 | recovery, cancel := CleanupContext(ctx, 10*time.Second) |
| 228 | defer cancel() |
| 229 | fresh, err := hostclaim.Read(recovery, d.Claims, o.Host) |
| 230 | if err != nil { |
| 231 | return err |
| 232 | } |
| 233 | rr := retryOf(fresh) |
| 234 | if rr == nil || rr.EpisodeID != episode || rr.Attempt == nil || *rr.Attempt != attempt { |
| 235 | return hostbox.ErrChanged |
| 236 | } |
| 237 | return d.settle(recovery, fresh, s.RetryClaimWrite) |
| 238 | } |
| 239 | s.Healthy = func(ctx context.Context, _ int, cmd *exec.Cmd) error { |
| 240 | if cmd == nil || cmd.Process == nil { |
| 241 | return hostbox.ErrChanged |
| 242 | } |
| 243 | if err := d.ExactHealth(ctx, cmd, attempt); err != nil { |
| 244 | return err |
| 245 | } |
| 246 | inputs, err := d.Inputs(ctx, rt) |
| 247 | if err != nil { |
| 248 | return err |
| 249 | } |
| 250 | if inputs != attempt.LaunchInputs { |
| 251 | return hostbox.ErrChanged |
| 252 | } |
| 253 | fresh, err := hostclaim.Read(ctx, d.Claims, o.Host) |
| 254 | if err != nil { |
| 255 | return err |
| 256 | } |
| 257 | if !sameAdmission(fresh, o) { |
| 258 | return hostbox.ErrChanged |
| 259 | } |
| 260 | _, err = hostclaim.Update(ctx, d.Claims, fresh, hostclaim.ClearWakeRetryHealthy(hostclaim.WakeRetryHealth{WakeRetryResult: hostclaim.WakeRetryResult{HolderIncarnation: inc, EpisodeID: episode, Attempt: attempt, Now: d.Now()}, ReadinessPassed: true})) |
| 261 | return err |
| 262 | } |
| 263 | return s, nil |
| 264 | } |
| 265 | return &out, nil |
| 266 | } |
| 267 | |
| 268 | // SettleAbandoned is the seed rerun's settlement boundary. Its caller holds the |
| 269 | // host exclusion. It spends no new attempt and remains usable below disk floor. |
| 270 | func (d ServingRetries) SettleAbandoned(ctx context.Context, expected hostclaim.Observed) (hostclaim.Observed, error) { |
| 271 | if !d.valid() || ctx == nil || expected.Absent || expected.Claim == nil || expected.Claim.Holder == nil || expected.Claim.Op.Kind != hostclaim.KindServing || expected.Claim.Op.Exec != d.Box { |
| 272 | return hostclaim.Observed{}, hostwrite.ErrUnavailable |
| 273 | } |
| 274 | fresh, err := hostclaim.Read(ctx, d.Claims, expected.Host) |
| 275 | if err != nil { |
| 276 | return hostclaim.Observed{}, err |
| 277 | } |
| 278 | if fresh.Absent || fresh.ETag != expected.ETag || !hostclaim.SameWriterIdentity(fresh, expected) { |
| 279 | return hostclaim.Observed{}, hostbox.ErrChanged |
| 280 | } |
| 281 | if err := d.settle(ctx, fresh); err != nil { |
| 282 | return hostclaim.Observed{}, err |
| 283 | } |
| 284 | return hostclaim.Read(ctx, d.Claims, expected.Host) |
| 285 | } |
| 286 | |
amadan