| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | |
| 7 | "github.com/carlosframework/platform/internal/hostwrite" |
| 8 | ) |
| 9 | |
| 10 | var ( |
| 11 | errSwapNotAwake = errors.New("swap: instance not awake") |
| 12 | errSwapLeaseLost = errors.New("swap: lease lost") |
| 13 | errSwapFenced = errors.New("swap: a stop-evidence operation holds the host") |
| 14 | errSwapShutdown = errors.New("swap: agent shutting down") |
| 15 | errSwapSuperseded = errors.New("swap: superseded by a newer attempt") |
| 16 | errSwapStopped = errors.New("swap: an intentional stop cancelled the attempt") |
| 17 | ) |
| 18 | |
| 19 | // selectWork picks the attempt's lifecycle work through hostwrite, never |
| 20 | // hostguard directly (spec §7.4). Claims off gives the legacy work. Claims |
| 21 | // on gives the claim-backed work, and a selector that cannot answer is a |
| 22 | // refusal, never the legacy work. |
| 23 | func (a *Service) selectWork(ctx context.Context, host string) (hostwrite.Work, error) { |
| 24 | work, err := a.cfg.HostWrites.Select(ctx, host, "spawn") |
| 25 | if err != nil { |
| 26 | return nil, err |
| 27 | } |
| 28 | if work == nil { |
| 29 | return nil, hostwrite.ErrUnavailable |
| 30 | } |
| 31 | return work, nil |
| 32 | } |
| 33 | |
| 34 | // admit runs fn as a §7.4 re-check. Claims off (hostwrite.Legacy): fn |
| 35 | // alone. fn takes a.mu itself, so the check is the entry lock against the |
| 36 | // barrier, the lease and the local fences that maintenance and moves set. |
| 37 | // Claims on: fn runs inside the selected work's With, so the claim and the |
| 38 | // installation stamp are validated first. A refusal there is returned as |
| 39 | // is and is never retried down the claims-off path. |
| 40 | func (a *Service) admit(ctx context.Context, work hostwrite.Work, fn func() error) error { |
| 41 | if work == nil { |
| 42 | return hostwrite.ErrUnavailable |
| 43 | } |
| 44 | if hostwrite.Legacy(work) { |
| 45 | return fn() |
| 46 | } |
| 47 | return work.With(ctx, hostwrite.Identity{}, fn) |
| 48 | } |
| 49 | |
| 50 | // swapChecksLocked is the four §7.4 checks on the entry the attempt targets: |
| 51 | // the entry is still awake, the lease is still held, no stop-evidence |
| 52 | // operation (the local fence) is in progress, and the attempt is still the |
| 53 | // newest. The agent's own shutdown counts as a stop in progress. newest is |
| 54 | // called with a.mu held. The caller holds a.mu. |
| 55 | func (a *Service) swapChecksLocked(e *actEntry, at Attempt, newest func(string) uint64) error { |
| 56 | return a.attemptCurrentLocked(e, at, newest, false) |
| 57 | } |
| 58 | |
| 59 | // attemptCurrentLocked is THE attempt gate (spec §6.3, §7.4): every exec |
| 60 | // point and every completion path of every attempt asks it, through |
| 61 | // swapChecksLocked, memberCurrentLocked, a hold's pinCheck, a claims-on |
| 62 | // spawn's before hook and a socketless respawn's attemptGate. waking is a |
| 63 | // hold's pinned wake, whose entry is still waking and holds no lease yet: |
| 64 | // the entry must be the host's, and the agent, the fences and the revision |
| 65 | // are checked. Otherwise the entry must be awake with its lease. The caller |
| 66 | // holds a.mu. |
| 67 | func (a *Service) attemptCurrentLocked(e *actEntry, at Attempt, newest func(string) uint64, waking bool) error { |
| 68 | if e == nil || a.entries[at.Host] != e { |
| 69 | return errSwapNotAwake |
| 70 | } |
| 71 | if err := a.ownedLocked(e, waking); err != nil { |
| 72 | return err |
| 73 | } |
| 74 | if newest(at.Host) != at.SchedRev { |
| 75 | return errSwapSuperseded |
| 76 | } |
| 77 | return nil |
| 78 | } |
| 79 | |
| 80 | // ownedLocked is the gate's OWNERSHIP check, apart from supersession (spec |
| 81 | // §7.4). Every exec asks it, an attempt's or not: a wake's and its |
| 82 | // fallback's, a worker respawn's (a crash restart, an attempt, its fallback, |
| 83 | // its restore) and a legacy sidecar's. Its refusals (errSwapNotAwake, |
| 84 | // errSwapLeaseLost, errSwapFenced, errSwapShutdown) are all gateNoOwnership: |
| 85 | // nothing spawns, a restore included. waking is an entry still waking, |
| 86 | // whose lease the wake holds itself (doWake's lease, not e.rep): the entry |
| 87 | // must be the host's, with no fence and no shutdown. The caller holds a.mu. |
| 88 | func (a *Service) ownedLocked(e *actEntry, waking bool) error { |
| 89 | switch { |
| 90 | case e == nil || a.entries[e.host] != e: |
| 91 | return errSwapNotAwake |
| 92 | case !waking && e.state != actAwake: // swapsim:M5 |
| 93 | return errSwapNotAwake |
| 94 | case !waking && (e.rep == nil || e.rep.Err() != nil): |
| 95 | return errSwapLeaseLost |
| 96 | case a.fences[e.host] != 0: |
| 97 | return errSwapFenced |
| 98 | case a.shuttingDown: |
| 99 | return errSwapShutdown |
| 100 | } |
| 101 | return nil |
| 102 | } |
| 103 | |
| 104 | // memberCurrentLocked is the gate for an attempt with a member in the |
| 105 | // operation set (an overlap's successor): the attempt is current, and its |
| 106 | // member is still in the set and not cancelled by the barrier. The caller |
| 107 | // holds a.mu. |
| 108 | func (a *Service) memberCurrentLocked(e *actEntry, m *genMember, mctx context.Context, at Attempt, newest func(string) uint64) error { |
| 109 | if err := a.attemptCurrentLocked(e, at, newest, false); err != nil { |
| 110 | return err |
| 111 | } |
| 112 | if e.ops.members[m.gen.N] != m || mctx.Err() != nil { |
| 113 | return errSwapStopped |
| 114 | } |
| 115 | return nil |
| 116 | } |
| 117 | |
amadan