| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "slices" |
| 8 | "time" |
| 9 | |
| 10 | "github.com/carlosframework/platform/internal/hostbox" |
| 11 | "github.com/carlosframework/platform/internal/hostclaim" |
| 12 | "github.com/carlosframework/platform/internal/hostguard" |
| 13 | "github.com/carlosframework/platform/internal/hostwrite" |
| 14 | "github.com/carlosframework/platform/internal/stopevidence" |
| 15 | ) |
| 16 | |
| 17 | // ErrClaimWritersStopped refuses new writer work on an attempt whose |
| 18 | // writers StopWriters has stopped. |
| 19 | var ErrClaimWritersStopped = errors.New("host claims: the session's writers are stopped") |
| 20 | |
| 21 | // The producers a writers-only stop keeps running: the session's |
| 22 | // replicator, its generation marker publisher and region HA's files |
| 23 | // publisher (claim §6.2a, L13). Every other producer is a writer: a role |
| 24 | // launch, or a timer delivery ("timer-" ids, ClaimTimerDelivery). |
| 25 | const ( |
| 26 | ClaimProducerReplicator = "replicator" |
| 27 | ClaimProducerPublisher = "publisher" |
| 28 | ClaimProducerFiles = "files" |
| 29 | ) |
| 30 | |
| 31 | func claimRetainedProducer(id string) bool { |
| 32 | return id == ClaimProducerReplicator || id == ClaimProducerPublisher || id == ClaimProducerFiles |
| 33 | } |
| 34 | |
| 35 | // ClaimWritersDrained is StopWriters' witness for one exact writer session |
| 36 | // (claim Plan 11, R2). It is no stop evidence: nothing is fenced, signed |
| 37 | // or published, and the claim is untouched. Root's writers-stopped |
| 38 | // observation (R3) and root's full stop still follow. |
| 39 | type ClaimWritersDrained struct { |
| 40 | // Target is the session's launch tuple: host, incarnation, stamp, |
| 41 | // writer session, box, enrollment, boot and controller. |
| 42 | Target hostguard.FenceTarget |
| 43 | Generation string |
| 44 | WriterBox string |
| 45 | // FenceClosed: the attempt and its producers admit no new writer. |
| 46 | FenceClosed bool |
| 47 | // LaunchJoined: every role launch producer was cancelled and joined. |
| 48 | LaunchJoined bool |
| 49 | // WorkDrained: every in-flight timer delivery was cancelled and joined. |
| 50 | WorkDrained bool |
| 51 | // SpawnsSettled: the launch journal holds no unsettled reservation. |
| 52 | SpawnsSettled bool |
| 53 | // CgroupEmpty: the session's containment was emptied and reads empty. |
| 54 | CgroupEmpty bool |
| 55 | // Cgroup is the containment's path relative to the cgroup root. |
| 56 | Cgroup string |
| 57 | DBPaths []string |
| 58 | // Drained and Retained are the producer ids joined and still running. |
| 59 | Drained, Retained []string |
| 60 | CheckedAt time.Time |
| 61 | } |
| 62 | |
| 63 | // Valid reports a complete witness. |
| 64 | func (d ClaimWritersDrained) Valid() bool { |
| 65 | t := d.Target |
| 66 | return t.Host != "" && t.Incarnation != "" && t.Stamp != 0 && t.Session != 0 && t.Box != "" && t.Enrollment != "" && t.BootID != "" && t.ControllerID != "" && |
| 67 | d.FenceClosed && d.LaunchJoined && d.WorkDrained && d.SpawnsSettled && d.CgroupEmpty && d.Cgroup != "" && len(d.DBPaths) != 0 && !d.CheckedAt.IsZero() |
| 68 | } |
| 69 | |
| 70 | // For checks the witness still names o's writer session: the same host, |
| 71 | // holder, stamp, session, generation and writer box. Any other session, |
| 72 | // including a later one on this box, gets nothing from it. |
| 73 | func (d ClaimWritersDrained) For(o hostclaim.Observed) error { |
| 74 | if !d.Valid() || o.Claim == nil || o.Claim.Holder == nil { |
| 75 | return hostbox.ErrChanged |
| 76 | } |
| 77 | h := o.Claim.Holder |
| 78 | if o.Host != d.Target.Host || h.Incarnation != d.Target.Incarnation || h.Stamp != d.Target.Stamp || h.Session != d.Target.Session || h.Generation != d.Generation || h.WriterBox != d.WriterBox { |
| 79 | return hostbox.ErrChanged |
| 80 | } |
| 81 | return nil |
| 82 | } |
| 83 | |
| 84 | // claimWritersJournal is the launch journal as StopWriters uses it. |
| 85 | type claimWritersJournal interface { |
| 86 | CloseWriters(context.Context, hostguard.FenceTarget) error |
| 87 | Read(context.Context, hostguard.FenceTarget) (hostbox.LaunchInventory, error) |
| 88 | Settled(context.Context, hostbox.LaunchReservation) error |
| 89 | } |
| 90 | |
| 91 | // claimWritersCgroup is the session's containment (the launch's parent). |
| 92 | type claimWritersCgroup interface { |
| 93 | Populated() (bool, error) |
| 94 | KillAndEmpty(context.Context) error |
| 95 | } |
| 96 | |
| 97 | // admitsWriters refuses once the attempt's writers are stopped. Every new |
| 98 | // launch, reservation, restore and producer of the attempt asks it. |
| 99 | func (a *claimWakeAttempt) admitsWriters() error { |
| 100 | a.mu.Lock() |
| 101 | defer a.mu.Unlock() |
| 102 | if a.writersClosed { |
| 103 | return ErrClaimWritersStopped |
| 104 | } |
| 105 | return nil |
| 106 | } |
| 107 | |
| 108 | // StopWriters stops the attempt's writers and keeps its replicator (claim |
| 109 | // Plan 11, R2). In order: close admission; cancel and join the role launch |
| 110 | // and the schedule runner's in-flight deliveries for the host (both are |
| 111 | // producers of the session); empty the session's containment; settle every |
| 112 | // spawn reservation the journal still holds; check the containment reads |
| 113 | // empty. The replicator, the generation publisher and the files publisher |
| 114 | // keep running, and the lease stays held. It never releases the claim or |
| 115 | // publishes evidence; root's stop does that. Once it has succeeded it |
| 116 | // returns the same witness again without repeating any effect; after a |
| 117 | // failure the next call resumes, each step being idempotent. |
| 118 | func (a *claimWakeAttempt) StopWriters(ctx context.Context) (ClaimWritersDrained, error) { |
| 119 | if a == nil || ctx == nil { |
| 120 | return ClaimWritersDrained{}, hostwrite.ErrUnavailable |
| 121 | } |
| 122 | a.writersMu.Lock() |
| 123 | defer a.writersMu.Unlock() |
| 124 | a.mu.Lock() |
| 125 | if a.writers != nil { |
| 126 | done := a.writers.clone() |
| 127 | a.mu.Unlock() |
| 128 | return done, nil |
| 129 | } |
| 130 | observed, in, owner := a.observed, a.inputs, a.owner |
| 131 | launch := a.launch |
| 132 | if launch == nil { |
| 133 | launch = a.prelaunch |
| 134 | } |
| 135 | journal, group := a.writersJournal, a.writersCgroup |
| 136 | a.mu.Unlock() |
| 137 | if launch == nil || owner == nil || owner.Producers == nil || journal == nil || group == nil || in.DBPath == "" { |
| 138 | return ClaimWritersDrained{}, hostwrite.ErrUnavailable |
| 139 | } |
| 140 | target := launch.Target |
| 141 | if observed.Claim == nil || observed.Claim.Holder == nil { |
| 142 | return ClaimWritersDrained{}, hostbox.ErrChanged |
| 143 | } |
| 144 | h := observed.Claim.Holder |
| 145 | if observed.Host != target.Host || h.Incarnation != target.Incarnation || h.Stamp != target.Stamp || h.Session != target.Session || target.Session == 0 { |
| 146 | return ClaimWritersDrained{}, hostbox.ErrChanged |
| 147 | } |
| 148 | // 1. Admission: the attempt refuses new launches and restores, and the |
| 149 | // session's producers refuse every writer from here on. |
| 150 | a.mu.Lock() |
| 151 | a.writersClosed = true |
| 152 | a.mu.Unlock() |
| 153 | producers, err := owner.Producers(ctx, observed) |
| 154 | if err != nil { |
| 155 | return ClaimWritersDrained{}, err |
| 156 | } |
| 157 | if err := journal.CloseWriters(ctx, target); err != nil { |
| 158 | return ClaimWritersDrained{}, err |
| 159 | } |
| 160 | // 2. The role launch and the timer deliveries, joined. |
| 161 | drained, err := producers.StopWriters(ctx, target) |
| 162 | if err != nil { |
| 163 | return ClaimWritersDrained{}, err |
| 164 | } |
| 165 | // 3. The containment: whatever the roles left is killed and waited for. |
| 166 | if err := group.KillAndEmpty(ctx); err != nil { |
| 167 | return ClaimWritersDrained{}, err |
| 168 | } |
| 169 | // 4. Spawn reservations: with the containment empty, nothing reserved |
| 170 | // or started can still run. An intended role is root's to resolve. |
| 171 | inventory, err := journal.Read(ctx, target) |
| 172 | if err != nil { |
| 173 | return ClaimWritersDrained{}, err |
| 174 | } |
| 175 | for _, r := range inventory.Reservations { |
| 176 | switch r.State { |
| 177 | case "settled": |
| 178 | case "reserved", "started": |
| 179 | if err := journal.Settled(ctx, r); err != nil { |
| 180 | return ClaimWritersDrained{}, err |
| 181 | } |
| 182 | default: |
| 183 | return ClaimWritersDrained{}, fmt.Errorf("%w: reservation %s is %s", ErrClaimStopInventory, r.ID, r.State) |
| 184 | } |
| 185 | } |
| 186 | settled, err := journal.Read(ctx, target) |
| 187 | if err != nil { |
| 188 | return ClaimWritersDrained{}, err |
| 189 | } |
| 190 | for _, r := range settled.Reservations { |
| 191 | if r.State != "settled" { |
| 192 | return ClaimWritersDrained{}, fmt.Errorf("%w: reservation %s is %s", ErrClaimStopInventory, r.ID, r.State) |
| 193 | } |
| 194 | } |
| 195 | // 5. The containment reads empty after the journal is settled. |
| 196 | populated, err := group.Populated() |
| 197 | if err != nil { |
| 198 | return ClaimWritersDrained{}, err |
| 199 | } |
| 200 | if populated { |
| 201 | return ClaimWritersDrained{}, fmt.Errorf("%w: the session's cgroup is still populated", ErrClaimStopInventory) |
| 202 | } |
| 203 | var retained []string |
| 204 | for _, id := range producers.Running() { |
| 205 | if !claimRetainedProducer(id) { |
| 206 | return ClaimWritersDrained{}, fmt.Errorf("%w: producer %s still runs", ErrClaimStopIdentity, id) |
| 207 | } |
| 208 | retained = append(retained, id) |
| 209 | } |
| 210 | // Every producer but the retained ones was joined in step 2: the role |
| 211 | // launch's and the timer deliveries' ("timer-" ids) alike. |
| 212 | done := ClaimWritersDrained{Target: target, Generation: h.Generation, WriterBox: h.WriterBox, FenceClosed: true, LaunchJoined: true, WorkDrained: true, SpawnsSettled: true, CgroupEmpty: true, |
| 213 | Cgroup: launch.ParentPath, DBPaths: []string{in.DBPath}, Drained: slices.Clip(drained), Retained: retained, CheckedAt: time.Now().UTC()} |
| 214 | if !done.Valid() { |
| 215 | return ClaimWritersDrained{}, hostwrite.ErrUnavailable |
| 216 | } |
| 217 | a.mu.Lock() |
| 218 | cached := done.clone() |
| 219 | a.writers = &cached |
| 220 | a.mu.Unlock() |
| 221 | return done, nil |
| 222 | } |
| 223 | |
| 224 | // recordReplicatorHints takes this process's own descriptors on the |
| 225 | // database files immediately before and after open, and keeps those that |
| 226 | // appeared: the replicator's descriptors (R3, the agent half). They are a |
| 227 | // narrowing hint for root, never evidence. |
| 228 | func recordReplicatorHints(ctx context.Context, procRoot string, pid int, db string, open func() error) ([]stopevidence.DescriptorHint, error) { |
| 229 | paths := []string{db, db + "-wal", db + "-shm"} |
| 230 | before, beforeErr := stopevidence.OwnDescriptors(ctx, procRoot, pid, paths) |
| 231 | if err := open(); err != nil { |
| 232 | return nil, err |
| 233 | } |
| 234 | if beforeErr != nil { |
| 235 | return nil, beforeErr |
| 236 | } |
| 237 | after, err := stopevidence.OwnDescriptors(ctx, procRoot, pid, paths) |
| 238 | if err != nil { |
| 239 | return nil, err |
| 240 | } |
| 241 | var hints []stopevidence.DescriptorHint |
| 242 | for _, d := range after { |
| 243 | if !slices.Contains(before, d) { |
| 244 | hints = append(hints, d) |
| 245 | } |
| 246 | } |
| 247 | return hints, nil |
| 248 | } |
| 249 | |
| 250 | func (d ClaimWritersDrained) clone() ClaimWritersDrained { |
| 251 | d.DBPaths = slices.Clone(d.DBPaths) |
| 252 | d.Drained = slices.Clone(d.Drained) |
| 253 | d.Retained = slices.Clone(d.Retained) |
| 254 | return d |
| 255 | } |
| 256 | |
amadan