| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/ecdh" |
| 6 | "crypto/ed25519" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "path/filepath" |
| 10 | "time" |
| 11 | |
| 12 | "github.com/carlosframework/platform/internal/activator" |
| 13 | "github.com/carlosframework/platform/internal/appconfig" |
| 14 | "github.com/carlosframework/platform/internal/controlsock" |
| 15 | "github.com/carlosframework/platform/internal/genselect" |
| 16 | "github.com/carlosframework/platform/internal/hostactors" |
| 17 | "github.com/carlosframework/platform/internal/hostbox" |
| 18 | "github.com/carlosframework/platform/internal/hostclaim" |
| 19 | execctl "github.com/carlosframework/platform/internal/hostexec" |
| 20 | "github.com/carlosframework/platform/internal/hostrelocate" |
| 21 | "github.com/carlosframework/platform/internal/hostwrite" |
| 22 | "github.com/carlosframework/platform/internal/lease" |
| 23 | "github.com/carlosframework/platform/internal/park" |
| 24 | "github.com/carlosframework/platform/internal/provision" |
| 25 | "github.com/carlosframework/platform/internal/registry" |
| 26 | "github.com/carlosframework/platform/internal/release" |
| 27 | ) |
| 28 | |
| 29 | // RelocationActivator is the part of activator.Service relocation uses: |
| 30 | // the probe attempt (Task 8d) and the local fence that ends a passed |
| 31 | // probe's session before the serving CAS (ruling 49). The fence is an |
| 32 | // owned one (ruling 75): relocation takes and meets it again only under |
| 33 | // the owner the probe stop persisted, and lifts it only under that owner. |
| 34 | // The activator's other paths do not ask for the owner: a plain |
| 35 | // FenceAndStop or LiftFence at the same generation, or a recovery probe |
| 36 | // opening on a RecoveryFence equal to it, replaces or lifts it and clears |
| 37 | // the owner, after which the owner can neither meet nor lift it. |
| 38 | type RelocationActivator interface { |
| 39 | ProbeWake(context.Context, activator.ProbeConfig) (activator.ProbeResult, error) |
| 40 | FenceAndStopAs(context.Context, string, uint64, string) (*activator.StopEvidence, error) |
| 41 | LiftFenceAs(string, uint64, string) bool |
| 42 | } |
| 43 | |
| 44 | var _ RelocationActivator = (*activator.Service)(nil) |
| 45 | |
| 46 | // RelocationAssemblyConfig is what the relocation providers need beyond |
| 47 | // the bound claim assembly. Task 10 fills it once per agent. |
| 48 | type RelocationAssemblyConfig struct { |
| 49 | Directory hostactors.Directory |
| 50 | // Data is the region's data bucket as a release.Store (files |
| 51 | // manifests), built with release.NewS3Store (ruling 50); Versions is |
| 52 | // the same bucket's versioned client (LTX, install manifests). |
| 53 | Data release.Store |
| 54 | Versions lease.VersionedStore |
| 55 | Bucket string |
| 56 | // Region is the logical placement region (genselect requests, pins, |
| 57 | // the worker). LeaseRegion is the AWS region lease.Acquire signs for: |
| 58 | // the roster's Cloud.Region, as main's Wake.LeaseRegion (ruling 51). |
| 59 | Region, LeaseRegion string |
| 60 | Registry *registry.Registry |
| 61 | DataDir string // hostguard.Guard.DataDir, the managed data root |
| 62 | OptDir string // the box's opt root: the version store and live/<host> |
| 63 | WorkDir string // the build's scratch: <DataDir>/relocate/build |
| 64 | ReleaseDir string // <DataDir>/relocate/release |
| 65 | SecretsKey *ecdh.PrivateKey |
| 66 | Activator RelocationActivator |
| 67 | // Source is the bound assembly's WorkSource (ClaimProviders.Selector): |
| 68 | // the install authority's guard, identity, Local and Prepare (ruling 46). |
| 69 | Source hostbox.WorkSource |
| 70 | // LaterProbeAttempts is set only once claim ask C11 is answered on |
| 71 | // main: root then authorizes a relocating launch at the admitted |
| 72 | // writer's stamp, not only at install_stamp (ruling 58). |
| 73 | LaterProbeAttempts bool |
| 74 | // Discharge is main's participation discharge for a completed |
| 75 | // relocation (productionClaimLocal.DischargeCompleted, N1). The hold |
| 76 | // and the activator fence are released only after it (ruling 66). |
| 77 | Discharge func(context.Context, hostclaim.Observed) error |
| 78 | Now func() time.Time |
| 79 | // ReleasePub and Ratchet are §4.1's evidence for the held stage |
| 80 | // (release.MaterializeHeldUnder): the agent's release key |
| 81 | // (CARLOS_RELEASE_PUBKEY) and the edge's pointer digest ratchet, |
| 82 | // pointerRatchet(false, optDir, dataDir, getenv), exactly as moves' T7 |
| 83 | // takes them in the agent (movespass.go, heldReleaseReady). Never |
| 84 | // root's: heldStagePass runs as root with pointerRatchet(true, ...). |
| 85 | ReleasePub ed25519.PublicKey |
| 86 | Ratchet *release.DigestRatchet |
| 87 | |
| 88 | // render, materialize and verifyLive are appconfig.RenderHeldUnder, |
| 89 | // release.MaterializeHeldUnder (with ReleasePub and Ratchet bound) and |
| 90 | // release.VerifyHeldLiveUnder; nil is those. Tests replace them. |
| 91 | render func(context.Context, release.Store, *ecdh.PrivateKey, string, registry.Route, string, string, func(appconfig.HeldRender) (hostwrite.Work, error)) (appconfig.HeldRender, error) |
| 92 | materialize func(context.Context, release.Store, string, registry.Route, string, *release.ProofGate, hostwrite.Work, func(func() error) error) (*release.Manifest, error) |
| 93 | verifyLive func(context.Context, release.Store, string, registry.Route, string, *release.ProofGate, func(func() error) error) error |
| 94 | } |
| 95 | |
| 96 | // RelocationEffects binds every provider to this bound assembly. ack is |
| 97 | // main's participation owner (productionClaimLocal.Acknowledged). |
| 98 | func (a ClaimAssembly) RelocationEffects(cfg RelocationAssemblyConfig, ack func(context.Context, hostclaim.Observed) (execctl.State, error)) *RelocationEffects { |
| 99 | if cfg.Now == nil { |
| 100 | cfg.Now = time.Now |
| 101 | } |
| 102 | receipts := RelocationReleaseReceipts{Store: a.Claims, Directory: cfg.Directory} |
| 103 | paths := relocationHostPaths(cfg.DataDir, cfg.Registry) |
| 104 | return &RelocationEffects{ |
| 105 | Claims: a.Claims, |
| 106 | Participation: ack, |
| 107 | Identity: func(ctx context.Context) (RelocationIdentity, error) { return a.localStopIdentity(ctx) }, |
| 108 | Producers: func(ctx context.Context, o hostclaim.Observed) (StopDrainer, error) { |
| 109 | if a.Wake == nil || a.Wake.Producers == nil { |
| 110 | return nil, ErrClaimStopUnavailable |
| 111 | } |
| 112 | p, err := a.Wake.Producers(ctx, o) |
| 113 | if err != nil { |
| 114 | return nil, err |
| 115 | } |
| 116 | return p, nil |
| 117 | }, |
| 118 | Collect: func(ctx context.Context, r StopRequest, d StopDrainer) (StopResult, error) { |
| 119 | if _, ok := d.(RecoveryReleaseDrain); ok { |
| 120 | r.Recover = true // 6b's authenticated recovery path: root replays the durable receipt |
| 121 | } |
| 122 | if a.collectStop != nil { |
| 123 | return a.collectStop(ctx, r, d) |
| 124 | } |
| 125 | return CollectClaimStop(ctx, controlsock.Path, r, d) |
| 126 | }, |
| 127 | WritersStopped: func(ctx context.Context, o hostclaim.Observed) (ClaimWritersDrained, WritersStoppedEvidence, error) { |
| 128 | return a.WritersStopped(ctx, o.Host) |
| 129 | }, |
| 130 | Releases: ReleaseObligation{Receipts: receipts}, |
| 131 | FlushDB: a.relocationFlushDB(cfg), |
| 132 | FlushFiles: ReleaseFlushFiles{Data: cfg.Data, Registry: cfg.Registry, |
| 133 | FilesDir: func(host string) string { return paths(host).FilesDir }, EnvPath: func(host string) string { return paths(host).EnvPath }}.Flush, |
| 134 | Positions: ReleasePositionsFile{Dir: cfg.ReleaseDir}, |
| 135 | PreflightKind: a.relocationPreflight(cfg), |
| 136 | BuildP: a.relocationBuild(cfg), |
| 137 | Install: a.relocationInstall(cfg), |
| 138 | VerifyLocal: a.relocationVerifyLocal(cfg), |
| 139 | Receipt: a.relocationReceipt(cfg), |
| 140 | Admit: a.relocationAdmit(cfg), |
| 141 | Passed: relocationPassedProbe(cfg.DataDir), |
| 142 | StopProbe: a.relocationStopProbe(cfg), |
| 143 | ReleaseHold: a.relocationReleaseHold(cfg), |
| 144 | } |
| 145 | } |
| 146 | |
| 147 | // relocationPreflight is ruling 48's gate before an attempt is counted: |
| 148 | // root speaks the isolated launch protocol, and its capability for kind |
| 149 | // relocating answers only with the attempt reader installed (Task 8d). |
| 150 | func (a ClaimAssembly) relocationPreflight(cfg RelocationAssemblyConfig) func(context.Context, hostclaim.Observed) error { |
| 151 | return func(ctx context.Context, o hostclaim.Observed) error { |
| 152 | if a.Wake == nil || a.Wake.Roles == nil || o.Claim == nil { |
| 153 | return activator.ErrClaimIsolationUnavailable |
| 154 | } |
| 155 | // Ruling 58: main's root pins every relocating launch to |
| 156 | // install_stamp (claimLaunchAuthority), and a cold attempt after |
| 157 | // P's opening runs at a new stamp. Until claim ask C11 relaxes the |
| 158 | // pin to P's opening, such an attempt is not made. |
| 159 | if hostclaim.PreparedOpened(o.Claim) && !cfg.LaterProbeAttempts { |
| 160 | return hostrelocate.ErrLaterAttemptUnlaunchable |
| 161 | } |
| 162 | // A defect in the claim the attempt would be made over (the |
| 163 | // release, the prepared authority, the pin, probing_since) is |
| 164 | // deterministic: it is found here, before the attempt is counted, |
| 165 | // never after three counted refusals (review M1). |
| 166 | if _, err := relocationProbeClaim(o.Claim); err != nil { |
| 167 | return err |
| 168 | } |
| 169 | if err := a.Wake.Roles.Preflight(ctx, ClaimLaunchInputs{}); err != nil { |
| 170 | return err |
| 171 | } |
| 172 | return a.Wake.Roles.PreflightKind(ctx, hostclaim.KindRelocating) |
| 173 | } |
| 174 | } |
| 175 | |
| 176 | // relocationPaths are one host's files directory and env file, derived |
| 177 | // from its route by the layout the activator would classify it with |
| 178 | // (design §3.1: a nested row's files live in its instance directory with |
| 179 | // the same names). Every relocation target route is nested (ruling 45). |
| 180 | type relocationPaths struct{ FilesDir, EnvPath string } |
| 181 | |
| 182 | func relocationPathsFor(dataDir string, rt registry.Route) relocationPaths { |
| 183 | dir := dataDir |
| 184 | if layout, _ := activator.Classify(rt, activator.LayoutInputs{DataDir: dataDir}); layout == activator.LayoutNested { |
| 185 | dir = provision.InstanceDir(dataDir, rt.Host) |
| 186 | } |
| 187 | return relocationPaths{FilesDir: filepath.Join(dir, rt.Host), EnvPath: filepath.Join(dir, rt.Host+".env")} |
| 188 | } |
| 189 | |
| 190 | // relocationHostPaths reads the host's own row: the source's flush |
| 191 | // publishes the paths its writer used. No row, or a row that cannot be |
| 192 | // read, is no paths, and ReleaseFlushFiles.Flush refuses empty paths |
| 193 | // before it stages anything rather than guess. |
| 194 | func relocationHostPaths(dataDir string, reg *registry.Registry) func(string) relocationPaths { |
| 195 | return func(host string) relocationPaths { |
| 196 | if reg == nil { |
| 197 | return relocationPaths{} |
| 198 | } |
| 199 | rt, err := reg.Get(host) |
| 200 | if err != nil || rt == nil { |
| 201 | return relocationPaths{} |
| 202 | } |
| 203 | return relocationPathsFor(dataDir, *rt) |
| 204 | } |
| 205 | } |
| 206 | |
| 207 | // relocationFlushDB is release step 2 (ruling 8) for the session's own |
| 208 | // replicator: releaseFinalTXID over the claim wake's attempt for the host. |
| 209 | func (a ClaimAssembly) relocationFlushDB(cfg RelocationAssemblyConfig) func(context.Context, hostclaim.Observed) (uint64, error) { |
| 210 | return func(ctx context.Context, o hostclaim.Observed) (uint64, error) { |
| 211 | if a.Wake == nil || o.Claim == nil || o.Claim.Holder == nil { |
| 212 | return 0, ErrClaimStopUnavailable |
| 213 | } |
| 214 | att := a.Wake.get(o.Host) |
| 215 | if att == nil || att.rep == nil || cfg.Versions == nil || cfg.Bucket == "" { |
| 216 | return 0, ErrClaimStopUnavailable |
| 217 | } |
| 218 | sel := func(ctx context.Context, r genselect.Request) (*genselect.Plan, error) { |
| 219 | return genselect.Select(ctx, cfg.Versions, r) |
| 220 | } |
| 221 | return releaseFinalTXID(ctx, att.rep, sel, genselect.Request{Bucket: cfg.Bucket, Region: cfg.Region, Host: o.Host, Holder: park.HolderRegionView(*o.Claim.Holder, cfg.Source.HomeRegion, cfg.Region)}) |
| 222 | } |
| 223 | } |
| 224 | |
| 225 | // releaseReplica is the part of the session's replicator the release's |
| 226 | // database flush uses (lease.Replicator). |
| 227 | type releaseReplica interface { |
| 228 | SyncAndWait(context.Context) error |
| 229 | } |
| 230 | |
| 231 | // releaseFinalTXID is release step 2 (ruling 8): SyncAndWait with every |
| 232 | // writer stopped, then N is the newest level-0 MaxTXID, which must equal |
| 233 | // the plan's MaxTXID. |
| 234 | // |
| 235 | // An empty listing is not evidence of an empty database (fix round 1, I2; |
| 236 | // N1): a wrong bucket, or a listing not yet visible, reads the same. N is |
| 237 | // 0 only for writer session 0, a holder no writer was ever admitted to |
| 238 | // (pin.go's "session-0", ruling 25). An empty listing for any other |
| 239 | // session refuses: the release is never journaled, and the worker marks |
| 240 | // the op stuck ReleaseDeadline after releasing began, for an operator, as |
| 241 | // a refused pin leaves it. |
| 242 | func releaseFinalTXID(ctx context.Context, rep releaseReplica, sel func(context.Context, genselect.Request) (*genselect.Plan, error), req genselect.Request) (uint64, error) { |
| 243 | if rep == nil || sel == nil { |
| 244 | return 0, ErrClaimStopUnavailable |
| 245 | } |
| 246 | if err := rep.SyncAndWait(ctx); err != nil { |
| 247 | return 0, err |
| 248 | } |
| 249 | plan, err := sel(ctx, req) |
| 250 | if errors.Is(err, genselect.ErrEmpty) { |
| 251 | if req.Holder.Session == 0 { |
| 252 | return 0, nil |
| 253 | } |
| 254 | return 0, fmt.Errorf("%w: the listing is empty for writer session %d", ErrClaimStopUnavailable, req.Holder.Session) |
| 255 | } |
| 256 | if err != nil { |
| 257 | return 0, err |
| 258 | } |
| 259 | if plan == nil { |
| 260 | return 0, ErrClaimStopUnavailable |
| 261 | } |
| 262 | if plan.MaxLevel0TXID != plan.MaxTXID { |
| 263 | return 0, fmt.Errorf("%w: level 0 at %d, the database at %d", ErrClaimStopUnavailable, plan.MaxLevel0TXID, plan.MaxTXID) |
| 264 | } |
| 265 | return plan.MaxTXID, nil |
| 266 | } |
| 267 | |
amadan