| 1 | package hostops |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "sync" |
| 7 | "sync/atomic" |
| 8 | |
| 9 | "github.com/carlosframework/platform/internal/hostclaim" |
| 10 | "github.com/carlosframework/platform/internal/provision" |
| 11 | "github.com/carlosframework/platform/internal/release" |
| 12 | ) |
| 13 | |
| 14 | // moveStartAction is the action of StartClaimMoveRecord's record effect, |
| 15 | // the only effect kind ResolveMoveStart settles. |
| 16 | const moveStartAction = "move-start" |
| 17 | |
| 18 | // Outcome is ResolveMoveStart's answer for one move-start effect. |
| 19 | type Outcome string |
| 20 | |
| 21 | const ( |
| 22 | // OutcomeComplete: the record-start effect applied. The journal retains |
| 23 | // the move's own outcome, including a later abort or rollback. |
| 24 | OutcomeComplete Outcome = "complete" |
| 25 | // OutcomeNotApplied: it does not, and no record Put the start issued |
| 26 | // can still land. |
| 27 | OutcomeNotApplied Outcome = "not-applied" |
| 28 | // OutcomeUncertain: neither is established; the effect stays uncertain |
| 29 | // and the error says why. |
| 30 | OutcomeUncertain Outcome = "uncertain" |
| 31 | ) |
| 32 | |
| 33 | // ErrMoveStartFenceUnproven retains uncertainty when capture is missing, |
| 34 | // the sequence regressed or repeated CAS conflicts prevented a proof. |
| 35 | var ErrMoveStartFenceUnproven = errors.New("hostops: move-start fence is unproven") |
| 36 | |
| 37 | // moveStartPuts binds every attempted conditional Put to its read sequence. |
| 38 | // Only the issuing handle holds it. Missing capture cannot prove a fence. |
| 39 | type moveStartPuts struct { |
| 40 | id, incarnation, key string |
| 41 | holder hostclaim.Holder |
| 42 | mu sync.Mutex |
| 43 | reads map[string]uint64 |
| 44 | bases []uint64 |
| 45 | } |
| 46 | |
| 47 | func (p *moveStartPuts) highestBase() (uint64, bool) { |
| 48 | p.mu.Lock() |
| 49 | defer p.mu.Unlock() |
| 50 | var high uint64 |
| 51 | for _, seq := range p.bases { |
| 52 | if seq > high { |
| 53 | high = seq |
| 54 | } |
| 55 | } |
| 56 | return high, len(p.bases) > 0 |
| 57 | } |
| 58 | |
| 59 | // moveStartSlot is a StartHandle's move-start Puts. StartClaimMoveRecord |
| 60 | // fills it from inside its issued effect, which runs at most once per |
| 61 | // handle: the handle's registration id is the move id, and so the effect id. |
| 62 | type moveStartSlot struct{ puts atomic.Pointer[moveStartPuts] } |
| 63 | |
| 64 | // moveStartRecorder is the store StartClaimMoveRecord gives the record |
| 65 | // writer. It notes the If-Match ETag of every Put of the move's record |
| 66 | // before passing the Put on. Every layer beneath (data boundary, conditional |
| 67 | // record store, raw bucket) either refuses the Put or issues it with that |
| 68 | // same ETag, so the noted set covers every Put that could still land. |
| 69 | type moveStartRecorder struct { |
| 70 | s release.Store |
| 71 | puts *moveStartPuts |
| 72 | capture func(context.Context, uint64) error |
| 73 | } |
| 74 | |
| 75 | var _ release.Store = moveStartRecorder{} |
| 76 | |
| 77 | func (r moveStartRecorder) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 78 | body, etag, err := r.s.Get(ctx, key) |
| 79 | if err == nil && key == r.puts.key { |
| 80 | _, seq, parseErr := recordSequence(body) |
| 81 | if parseErr != nil { |
| 82 | return nil, "", parseErr |
| 83 | } |
| 84 | if etag == "" { |
| 85 | return nil, "", ErrIdentity |
| 86 | } |
| 87 | r.puts.mu.Lock() |
| 88 | if r.puts.reads == nil { |
| 89 | r.puts.reads = map[string]uint64{} |
| 90 | } |
| 91 | r.puts.reads[etag] = seq |
| 92 | r.puts.mu.Unlock() |
| 93 | } |
| 94 | return body, etag, err |
| 95 | } |
| 96 | |
| 97 | func (r moveStartRecorder) Put(ctx context.Context, key string, body []byte, etag string) (string, error) { |
| 98 | if key == r.puts.key { |
| 99 | r.puts.mu.Lock() |
| 100 | seq, ok := r.puts.reads[etag] |
| 101 | if ok && etag != "" { |
| 102 | r.puts.bases = append(r.puts.bases, seq) |
| 103 | } |
| 104 | r.puts.mu.Unlock() |
| 105 | if !ok || etag == "" { |
| 106 | return "", ErrIdentity |
| 107 | } |
| 108 | if r.capture != nil { |
| 109 | if err := r.capture(ctx, seq); err != nil { |
| 110 | return "", err |
| 111 | } |
| 112 | } |
| 113 | } |
| 114 | return r.s.Put(ctx, key, body, etag) |
| 115 | } |
| 116 | |
| 117 | func (r moveStartRecorder) List(ctx context.Context, prefix string) ([]string, error) { |
| 118 | return r.s.List(ctx, prefix) |
| 119 | } |
| 120 | |
| 121 | func (r moveStartRecorder) ListDirs(ctx context.Context, prefix string) ([]string, error) { |
| 122 | return r.s.ListDirs(ctx, prefix) |
| 123 | } |
| 124 | |
| 125 | func (r moveStartRecorder) Delete(ctx context.Context, key string) error { |
| 126 | return r.s.Delete(ctx, key) |
| 127 | } |
| 128 | |
| 129 | // ResolveMoveStart resolves the exact issuing handle's uncertain move-start |
| 130 | // effect from its exact journal or a newer record sequence that fences every |
| 131 | // attempted Put. Only the runner records settlement; read errors and failed |
| 132 | // fence writes retain uncertainty. |
| 133 | // Previously settled results are returned without issuing any new write. |
| 134 | func ResolveMoveStart(ctx context.Context, h *StartHandle, o hostclaim.Observed, moveID string) (Outcome, error) { |
| 135 | if h == nil || h.w.ready() != nil || moveID == "" || moveID != h.id || o.Absent || o.Claim == nil || o.Claim.Holder == nil { |
| 136 | return OutcomeUncertain, ErrIdentity |
| 137 | } |
| 138 | c, holder := o.Claim, o.Claim.Holder |
| 139 | if c.Op.Kind != hostclaim.KindMoving || c.Op.ID != moveID || c.Op.Intent["admission_id"] != h.id || c.Op.Intent["app_id"] != h.appID || |
| 140 | holder.Family != "instance" || holder.Account != h.account || holder.App != h.app { |
| 141 | return OutcomeUncertain, ErrIdentity |
| 142 | } |
| 143 | puts := h.moveStart.puts.Load() |
| 144 | if puts == nil || puts.id != moveID || puts.incarnation != holder.Incarnation || puts.key != holder.Key { |
| 145 | return OutcomeUncertain, ErrIdentity |
| 146 | } |
| 147 | a, err := h.w.read(ctx, h.account, h.app, h.appID) |
| 148 | if err != nil { |
| 149 | return OutcomeUncertain, err |
| 150 | } |
| 151 | e, ok := a.Value.Effects[moveID] |
| 152 | if !ok || e.OperationID != h.id || e.AppID != h.appID || e.Owner != h.owner || e.Action != moveStartAction || e.Resource != holder.Key || |
| 153 | e.Inputs["move_id"] != moveID || e.Inputs["incarnation"] != holder.Incarnation { |
| 154 | return OutcomeUncertain, ErrIdentity |
| 155 | } |
| 156 | switch e.State { |
| 157 | case "complete": |
| 158 | return settledMoveStart(e, moveID) |
| 159 | case "uncertain": |
| 160 | default: |
| 161 | // Prepared or issued: the issuer has not joined and settled its call. |
| 162 | return OutcomeUncertain, ErrPending |
| 163 | } |
| 164 | result, err := decideMoveStart(ctx, h.w, puts, o) |
| 165 | if err != nil { |
| 166 | return OutcomeUncertain, err |
| 167 | } |
| 168 | if err = h.Runner(h.w.now).resolveUncertain(ctx, e, result); err != nil { |
| 169 | // A concurrent resolution may have recorded a decision first. |
| 170 | if a, rerr := h.w.read(ctx, h.account, h.app, h.appID); rerr == nil { |
| 171 | if cur, ok := a.Value.Effects[moveID]; ok && cur.State == "complete" { |
| 172 | return settledMoveStart(cur, moveID) |
| 173 | } |
| 174 | } |
| 175 | return OutcomeUncertain, err |
| 176 | } |
| 177 | if result == moveID { |
| 178 | return OutcomeComplete, nil |
| 179 | } |
| 180 | return OutcomeNotApplied, nil |
| 181 | } |
| 182 | |
| 183 | // settledMoveStart maps a complete move-start effect to its outcome. Any |
| 184 | // other recorded result is not a decision this function made or can read. |
| 185 | func settledMoveStart(e Effect, moveID string) (Outcome, error) { |
| 186 | switch e.ResultReference { |
| 187 | case moveID: |
| 188 | return OutcomeComplete, nil |
| 189 | case "not-applied": |
| 190 | return OutcomeNotApplied, nil |
| 191 | } |
| 192 | return OutcomeUncertain, ErrIdentity |
| 193 | } |
| 194 | |
| 195 | // decideMoveStart never uses ETag inequality as proof. A lost fence answer |
| 196 | // stays uncertain; a later call may prove that fence from its sequence. |
| 197 | func decideMoveStart(ctx context.Context, w *ControlWriter, puts *moveStartPuts, o hostclaim.Observed) (string, error) { |
| 198 | base, captured := puts.highestBase() |
| 199 | if !captured { |
| 200 | return "", ErrMoveStartFenceUnproven |
| 201 | } |
| 202 | for attempt := 0; attempt < 3; attempt++ { |
| 203 | current, err := ReadRecord(ctx, w.raw, puts.key) |
| 204 | if err != nil { |
| 205 | return "", err |
| 206 | } |
| 207 | if !matches(current, &puts.holder) { |
| 208 | return "", ErrIdentity |
| 209 | } |
| 210 | _, seq, err := recordSequence(current.Body) |
| 211 | if err != nil { |
| 212 | return "", err |
| 213 | } |
| 214 | rec, err := provision.DecodeRecord(current.Body) |
| 215 | if err != nil { |
| 216 | return "", err |
| 217 | } |
| 218 | if rec.Move != nil && rec.Move.ID == puts.id && rec.Move.Incarnation == puts.incarnation { |
| 219 | return puts.id, nil |
| 220 | } |
| 221 | // A box may finish the move while the console's start result is |
| 222 | // uncertain. The terminal journal on this verified incarnation is |
| 223 | // positive evidence too, even after Move has been cleared. |
| 224 | if rec.LastMove != nil && rec.LastMove.ID == puts.id { |
| 225 | return puts.id, nil |
| 226 | } |
| 227 | if seq > base { |
| 228 | return "not-applied", nil |
| 229 | } |
| 230 | // A lower counter is a violated migration/write invariant, not authority |
| 231 | // to repair it and settle. Equal requires a new conditional sequence. |
| 232 | if seq < base { |
| 233 | return "", ErrMoveStartFenceUnproven |
| 234 | } |
| 235 | if err := CheckCreateFreeze(ctx, w.raw, w.pub); err != nil { |
| 236 | return "", err |
| 237 | } |
| 238 | fresh, err := hostclaim.Read(ctx, w.raw, o.Host) |
| 239 | if err != nil { |
| 240 | return "", err |
| 241 | } |
| 242 | if fresh.Absent || fresh.ETag != o.ETag { |
| 243 | return "", hostclaim.ErrConflict |
| 244 | } |
| 245 | if _, _, err = provision.WriteRecord(provision.WithRecordSequence(ctx), w.raw, puts.key, current.Body, current.Body, current.ETag); err == nil { |
| 246 | return "not-applied", nil |
| 247 | } else if !errors.Is(err, release.ErrCASConflict) { |
| 248 | return "", err |
| 249 | } |
| 250 | // The delayed Put may have won: reread its exact journal before deciding. |
| 251 | } |
| 252 | return "", ErrMoveStartFenceUnproven |
| 253 | } |
| 254 | |
amadan