| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "io" |
| 8 | "log" |
| 9 | "os" |
| 10 | "sync/atomic" |
| 11 | "time" |
| 12 | |
| 13 | "github.com/carlosframework/platform/internal/activator" |
| 14 | "github.com/carlosframework/platform/internal/controlname" |
| 15 | "github.com/carlosframework/platform/internal/layout" |
| 16 | "github.com/carlosframework/platform/internal/provision" |
| 17 | "github.com/carlosframework/platform/internal/registry" |
| 18 | "github.com/carlosframework/platform/internal/release" |
| 19 | "github.com/carlosframework/platform/internal/seeds" |
| 20 | ) |
| 21 | |
| 22 | // storeSeedObjects reads seed snapshots from this box's data bucket. |
| 23 | type storeSeedObjects struct{ s release.Store } |
| 24 | |
| 25 | // Open streams seeds/<host>/<id>/snapshot.db. A store that cannot stream is |
| 26 | // refused rather than read whole: a seed can be gigabytes. |
| 27 | func (o storeSeedObjects) Open(ctx context.Context, host, id string) (io.ReadCloser, error) { |
| 28 | if err := seeds.Check(host, id); err != nil { |
| 29 | return nil, fmt.Errorf("seed %q of %s: %w", id, host, err) |
| 30 | } |
| 31 | op, ok := o.s.(release.Opener) |
| 32 | if !ok { |
| 33 | return nil, errors.New("the data bucket store cannot stream") |
| 34 | } |
| 35 | rc, _, err := op.Open(ctx, seeds.SnapshotKey(host, id)) |
| 36 | return rc, err |
| 37 | } |
| 38 | |
| 39 | // seedPlacedHook is ExecBackend.OnSeedPlaced (seeded-instances spec §3.4 |
| 40 | // step 5): the record's CAS to placed, then the delete of this seed's |
| 41 | // snapshot, never another's. The activator calls it only once the host's |
| 42 | // own replica holds the seed. |
| 43 | // |
| 44 | // It is called more than once for one placement: twice on a finishing wake |
| 45 | // (Restore's finishPlacement and doWake's step 5b), and by any box that |
| 46 | // wakes the host before the first call's delete lands. A record already |
| 47 | // placed (or committed) with this id is that case: a success, and the |
| 48 | // delete still runs, because the earlier call's delete may be the one that |
| 49 | // failed. A CAS refused for any other reason deletes nothing. |
| 50 | func seedPlacedHook(deploy, data release.Store) func(ctx context.Context, acct, app, host, id string) error { |
| 51 | return func(ctx context.Context, acct, app, host, id string) error { |
| 52 | if deploy == nil || data == nil { |
| 53 | return errors.New("seed recording is not wired on this box") |
| 54 | } |
| 55 | if err := seeds.Check(host, id); err != nil { |
| 56 | return fmt.Errorf("seed %q of %s: %w", id, host, err) |
| 57 | } |
| 58 | _, err := provision.SetRecordSeed(ctx, deploy, acct, app, host, provision.SeedPlace(id)) |
| 59 | if errors.Is(err, provision.ErrSeedConflict) { |
| 60 | rec, gerr := provision.GetRecord(ctx, deploy, acct, app, host) |
| 61 | if gerr != nil { |
| 62 | return gerr |
| 63 | } |
| 64 | if !placedWith(rec, id) { |
| 65 | return fmt.Errorf("seed %s of %s: %w", id, host, err) |
| 66 | } |
| 67 | } else if err != nil { |
| 68 | return err |
| 69 | } |
| 70 | if err := data.Delete(ctx, seeds.SnapshotKey(host, id)); err != nil && !errors.Is(err, os.ErrNotExist) { |
| 71 | return fmt.Errorf("delete seed %s of %s: %w", id, host, err) |
| 72 | } |
| 73 | return nil |
| 74 | } |
| 75 | } |
| 76 | |
| 77 | // placedWith reports whether rec's seed is id and has been placed: placed, |
| 78 | // or committed, which only follows placed. |
| 79 | func placedWith(rec *provision.Record, id string) bool { |
| 80 | return rec != nil && rec.Seed != nil && rec.Seed.ID == id && |
| 81 | (rec.Seed.State == seeds.StatePlaced || rec.Seed.State == seeds.StateCommitted) |
| 82 | } |
| 83 | |
| 84 | // armedSeedHooks is what reconcileSerialized hands the provision pass |
| 85 | // (ReconcileConfig.Seeds). Nil until the activator is armed with a data |
| 86 | // bucket: the pass then mints and finishes no seeded record. |
| 87 | var armedSeedHooks atomic.Pointer[provision.SeedHooks] |
| 88 | |
| 89 | // armSeedHooks arms the provision pass's seed hooks. Hooks without a data |
| 90 | // bucket are never armed: the pass's aborted-seed finish deletes through |
| 91 | // it. |
| 92 | func armSeedHooks(h *provision.SeedHooks) { |
| 93 | if h != nil && h.Data == nil { |
| 94 | h = nil |
| 95 | } |
| 96 | armedSeedHooks.Store(h) |
| 97 | } |
| 98 | |
| 99 | // seedDataOpener opens this region's data bucket, where seeds/ lives. |
| 100 | func seedDataOpener(bucket, region string) func() (release.Store, error) { |
| 101 | return func() (release.Store, error) { |
| 102 | s, err := release.NewS3StoreIn(context.Background(), bucket, region) |
| 103 | if err != nil { |
| 104 | return nil, err |
| 105 | } |
| 106 | return s, nil |
| 107 | } |
| 108 | } |
| 109 | |
| 110 | // seedBox is what the import swap and rollback need from this box beyond |
| 111 | // the two stores (preflight M7): the activator's fence, the key prefix of |
| 112 | // the replicas in the data bucket (cfg.Prefix), whether the startup unit |
| 113 | // sweep succeeded, and the fence generation source. |
| 114 | type seedBox struct { |
| 115 | Fence layout.Fence |
| 116 | Prefix string |
| 117 | Swept func() bool |
| 118 | NextFenceGen func() (uint64, error) |
| 119 | } |
| 120 | |
| 121 | // wireSeedPlacement arms seeded instances on this box (seeded-instances |
| 122 | // spec §3.4): the claim flag that makes the seed fallback refuse, the data |
| 123 | // bucket for snapshots, the record store for the placed CAS, and the |
| 124 | // provision pass's hooks. The claim flag is set whatever else happens. The |
| 125 | // rest is best-effort like the staging-fallback mirror, and all or nothing: |
| 126 | // a box that cannot build both stores places no seed, and a seeded row's |
| 127 | // wake fails instead of starting empty. |
| 128 | // |
| 129 | // The import swap's and rollback's hooks (spec §5.4, §5.6) are armed only |
| 130 | // with claim control off: a claim-armed box refuses seeds until #250, so it |
| 131 | // never swaps, and an import journaled before claims came on stays gated |
| 132 | // and untouched there. The rollback's seven hooks are armed together or not |
| 133 | // at all (provision runs no rollback missing one); without a fence from the |
| 134 | // activator none is. |
| 135 | // |
| 136 | // The rollback's fence is known by the generation its journal stores, so |
| 137 | // box.NextFenceGen must never repeat one across a restart |
| 138 | // (fenceGenerations). The hooks set no clock: the swap and the rollback |
| 139 | // read the pass's own (ReconcileConfig.Now). |
| 140 | func wireSeedPlacement(eb *activator.ExecBackend, claimArmed bool, openData, openDeploy func() (release.Store, error), box seedBox) { |
| 141 | eb.ClaimArmed = claimArmed |
| 142 | data, err := openData() |
| 143 | if err != nil { |
| 144 | log.Printf("agent: seed reader: %v (no seed can be placed on this box)", err) |
| 145 | return |
| 146 | } |
| 147 | deploy, err := openDeploy() |
| 148 | if err != nil { |
| 149 | log.Printf("agent: seed recorder: %v (no seed can be placed on this box)", err) |
| 150 | return |
| 151 | } |
| 152 | eb.Seeds = storeSeedObjects{data} |
| 153 | eb.OnSeedPlaced = seedPlacedHook(deploy, data) |
| 154 | h := &provision.SeedHooks{Data: data} |
| 155 | if !claimArmed { |
| 156 | h.NestedOK = eb.NestedOK |
| 157 | h.LegacyOwners = func(ctx context.Context) (*seeds.LegacyOwners, error) { |
| 158 | o, _, err := seeds.ReadLegacyOwners(ctx, deploy) |
| 159 | return o, err |
| 160 | } |
| 161 | h.EnvBefore = activator.SeedEnvBefore |
| 162 | if box.Fence != nil && box.Swept != nil && box.NextFenceGen != nil { |
| 163 | fence := box.Fence |
| 164 | h.FenceForRollback = fenceForRollback(fence, box.Swept) |
| 165 | h.LiftFence = func(host string, gen uint64) { fence.LiftFence(host, gen) } |
| 166 | h.RemoveLocal = func(rt registry.Route) error { return activator.PurgeSeedLocal(rt.DBPath) } |
| 167 | h.LocalSeedID = activator.ReadSeedMarker |
| 168 | h.RetireKey = activator.RetireSeedKey |
| 169 | h.DeleteReplica = deleteReplica(data, box.Prefix) |
| 170 | h.NextFenceGen = box.NextFenceGen |
| 171 | } |
| 172 | } |
| 173 | armSeedHooks(h) |
| 174 | } |
| 175 | |
| 176 | // seedSweptAtStart is whether the agent's startup unit sweep succeeded, |
| 177 | // for the import rollback's stop proof: the same fact the layout pass is |
| 178 | // built with. It stays false with claim control on, where the activator |
| 179 | // starts without a sweep (and no rollback is armed). |
| 180 | var seedSweptAtStart atomic.Bool |
| 181 | |
| 182 | // fenceGenerations is this box's fence generation source, the registry's: |
| 183 | // never 0, and never one it handed out before, across agent restarts. The |
| 184 | // layout pass draws from it too (newLayoutPass), so a rollback resumed |
| 185 | // after a restart can never meet the pass's fence under its own number. |
| 186 | func fenceGenerations(reg *registry.Registry) func() (uint64, error) { |
| 187 | return func() (uint64, error) { return reg.NextFenceGeneration(time.Now()) } |
| 188 | } |
| 189 | |
| 190 | // rollbackFenceTimeout bounds one fence-and-stop, as the layout pass's. |
| 191 | const rollbackFenceTimeout = 30 * time.Second |
| 192 | |
| 193 | // fenceForRollback is the import rollback's step 2 (seeded-instances spec |
| 194 | // §5.6), by internal/layout/pass.go's rule (Pass.quiesce) for the same |
| 195 | // activator API. A fence already holding gen is this rollback's own, from |
| 196 | // an earlier pass of this process: proceed. Otherwise fence and stop under |
| 197 | // a 30-second deadline. A proven stop proceeds; another operation's fence |
| 198 | // defers, touching nothing; an unproven stop proceeds only when the box |
| 199 | // was swept at start, the host is quiesced and its stop is not uncertain. |
| 200 | // Anything else lifts this attempt's fence and defers. |
| 201 | // |
| 202 | // Generation 0 is no fence's: FenceGeneration answers 0 for a host with |
| 203 | // none, which must never read as ours. |
| 204 | func fenceForRollback(f layout.Fence, swept func() bool) func(ctx context.Context, host string, gen uint64) (bool, error) { |
| 205 | return func(ctx context.Context, host string, gen uint64) (bool, error) { |
| 206 | if gen == 0 { |
| 207 | return false, activator.ErrFenceGeneration |
| 208 | } |
| 209 | if f.FenceGeneration(host) == gen { |
| 210 | return true, nil |
| 211 | } |
| 212 | fctx, cancel := context.WithTimeout(ctx, rollbackFenceTimeout) |
| 213 | defer cancel() |
| 214 | _, err := f.FenceAndStopIfUnfenced(fctx, host, gen) |
| 215 | switch { |
| 216 | case err == nil: |
| 217 | return true, nil |
| 218 | case errors.Is(err, activator.ErrFenceHeld): |
| 219 | return false, err |
| 220 | case errors.Is(err, activator.ErrStopUnproven) && swept() && f.Quiesced(host) && !f.StopUncertain(host): |
| 221 | return true, nil |
| 222 | } |
| 223 | f.LiftFence(host, gen) |
| 224 | return false, err |
| 225 | } |
| 226 | } |
| 227 | |
| 228 | // deleteReplica is the rollback's step 6 for the host's replica: every key |
| 229 | // under <prefix>/<host>/ in the region data bucket, the prefix the |
| 230 | // activator replicates to (activator.HibernatePrefix). The trailing slash |
| 231 | // keeps a host whose name only begins with this one out of it. Nothing |
| 232 | // there is not an error. |
| 233 | func deleteReplica(data release.Store, prefix string) func(ctx context.Context, host string) error { |
| 234 | return func(ctx context.Context, host string) error { |
| 235 | if !controlname.ValidHostShape(host) { |
| 236 | return fmt.Errorf("delete the replica of %q: not a host name", host) |
| 237 | } |
| 238 | keys, err := data.List(ctx, activator.HibernatePrefix(prefix, host)+"/") |
| 239 | if err != nil { |
| 240 | return err |
| 241 | } |
| 242 | for _, k := range keys { |
| 243 | if err := data.Delete(ctx, k); err != nil && !errors.Is(err, os.ErrNotExist) { |
| 244 | return err |
| 245 | } |
| 246 | } |
| 247 | return nil |
| 248 | } |
| 249 | } |
| 250 | |
amadan