amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package host
2
3import (
4 "context"
5 "crypto/ed25519"
6 "encoding/json"
7 "errors"
8 "fmt"
9 "io"
10 "log"
11 "net"
12 "net/http"
13 "os"
14 "path/filepath"
15 "slices"
16 "sync"
17 "time"
18
19 "github.com/carlosframework/platform/internal/hostbox"
20 "github.com/carlosframework/platform/internal/hostclaim"
21 "github.com/carlosframework/platform/internal/hostwrite"
22 "github.com/carlosframework/platform/internal/instancectl"
23 "github.com/carlosframework/platform/internal/moves"
24 "github.com/carlosframework/platform/internal/provision"
25 "github.com/carlosframework/platform/internal/regionname"
26 "github.com/carlosframework/platform/internal/registry"
27 "github.com/carlosframework/platform/internal/release"
28 "github.com/carlosframework/platform/internal/rootstate"
29 "github.com/carlosframework/platform/internal/safefile"
30)
31
32// heldReleaseReceiptDir is where root keeps its publish receipts: its own
33// claim state directory (carlos-instance-control's StateDirectory in claim
34// mode), which the agent cannot write.
35const heldReleaseReceiptDir = "/var/lib/carlos-claims/held-release"
36
37// movesHeldReleaseRoot is root's side of the held release (Task 7b; see
38// movesheldrelease.go). Every authority it checks comes from root's own
39// providers: the authenticated peer, the claim and record it reads from the
40// deployment store, the registry row, the render journal it reads through
41// the anchored path, and its own held-release selection. A request only
42// names what the agent expects them to say.
43type movesHeldReleaseRoot struct {
44 OpenRelease func() (rootstate.Store, error)
45 // Authenticate is root's peer authentication (the edge unit's main
46 // process), the claim root's own.
47 Authenticate func(context.Context, net.Conn) (StopPeer, error)
48 // Actor is this box's verified roster identity.
49 Actor func(context.Context) (hostbox.Actor, error)
50 // Store is the deployment store: releases and the source's report.
51 Store release.Store
52 Claims func(context.Context, string) (hostclaim.Observed, error)
53 Records MovesRecordReader
54 Route func(string) (*registry.Route, error)
55 // Works selects the claim's held-release work for host
56 // (hostbox.WorkSource.Select, built for actor); Selected reads back the
57 // selection it captured (hostbox.Selected).
58 Works func(ctx context.Context, actor hostbox.Actor, host, action string) (hostwrite.Work, error)
59 Selected func(hostwrite.Work) (hostbox.Selection, error)
60 // Pub, Ratchet and Proof are §4.1's and D3's evidence, as for root's
61 // adoption pass: the release key, root's pointer digest ratchet and a
62 // fresh proof gate per request.
63 Pub ed25519.PublicKey
64 Ratchet func() *release.DigestRatchet
65 Proof func() *release.ProofGate
66 OptDir string
67 generationPins release.PinSource
68 // DataDir is the box's data directory: the render journals and the
69 // held-stage lock.
70 DataDir string
71 ReceiptDir string
72
73 // Test seams. Nil is production.
74 stage func(context.Context, release.Store, ed25519.PublicKey, *release.DigestRatchet, string, registry.Route, string, *release.ProofGate) (*release.Manifest, error)
75 materialize func(context.Context, release.Store, ed25519.PublicKey, *release.DigestRatchet, string, registry.Route, string, *release.ProofGate, hostwrite.Work, func(func() error) error) (*release.Manifest, error)
76 renderHash func(dataDir, host, op string) (string, error)
77 afterPublish func(host string) error
78
79 // The bound (fix round 2, P3-4): one request per host at a time, and
80 // heldReleaseConcurrency in all.
81 mu sync.Mutex
82 inflight map[string]bool
83}
84
85// admit takes host's slot and one of the global ones, or refuses at once
86// with errHeldReleaseBusy: root never queues an agent's requests. The
87// returned func frees both.
88func (h *movesHeldReleaseRoot) admit(host string) (func(), error) {
89 h.mu.Lock()
90 defer h.mu.Unlock()
91 if h.inflight == nil {
92 h.inflight = map[string]bool{}
93 }
94 if h.inflight[host] || len(h.inflight) >= heldReleaseConcurrency {
95 return nil, errHeldReleaseBusy
96 }
97 h.inflight[host] = true
98 return func() {
99 h.mu.Lock()
100 delete(h.inflight, host)
101 h.mu.Unlock()
102 }, nil
103}
104
105func (h *movesHeldReleaseRoot) complete() bool {
106 return h != nil && h.OpenRelease != nil && h.Actor != nil && h.Store != nil && h.Claims != nil && h.Records != nil && h.Route != nil && h.Works != nil && h.Selected != nil &&
107 len(h.Pub) == ed25519.PublicKeySize && h.Ratchet != nil && h.Proof != nil && filepath.IsAbs(h.OptDir) && filepath.IsAbs(h.DataDir) && filepath.IsAbs(h.ReceiptDir)
108}
109
110// ServeHTTP answers POST /claim/held-release for an authenticated peer.
111// The peer is authenticated before the body is read.
112func (h *movesHeldReleaseRoot) ServeHTTP(w http.ResponseWriter, r *http.Request) {
113 if h == nil || h.Authenticate == nil || r.Method != http.MethodPost || r.URL.Path != heldReleasePath {
114 http.Error(w, "refused", http.StatusServiceUnavailable)
115 return
116 }
117 conn := instancectl.RequestConn(r.Context())
118 if conn == nil {
119 http.Error(w, "refused", http.StatusForbidden)
120 return
121 }
122 peer, err := h.Authenticate(r.Context(), conn)
123 if err != nil {
124 http.Error(w, "refused", http.StatusForbidden)
125 return
126 }
127 q, err := decodeHeldReleaseRequest(r.Body)
128 if err != nil {
129 http.Error(w, "refused", http.StatusBadRequest)
130 return
131 }
132 out, err := h.Serve(r.Context(), peer, q)
133 if err != nil {
134 // Fix round 2, P3-2: a request's fields reach the log only once they
135 // have passed validation, and then quoted, as is the error, so the
136 // agent cannot forge a line in instance-control's journal.
137 if q.validate() != nil {
138 log.Printf("moves: held release refused: a malformed request: %q", err.Error())
139 } else {
140 log.Printf("moves: %q: held release %q of %q refused: %q", q.Host, q.Action, q.Version, err.Error())
141 }
142 http.Error(w, "refused", heldReleaseStatus(err))
143 return
144 }
145 w.Header().Set("Content-Type", "application/json")
146 w.WriteHeader(http.StatusOK)
147 _ = json.NewEncoder(w).Encode(out)
148}
149
150// heldReleaseStatus is the status each refusal answers with; the agent's
151// client maps it back to the same typed error.
152func heldReleaseStatus(err error) int {
153 var refusal *MovesPolicyRefusal
154 switch {
155 case errors.Is(err, errHeldReleaseShape):
156 return http.StatusBadRequest
157 case errors.Is(err, ErrClaimStopIdentity):
158 return http.StatusForbidden
159 case errors.Is(err, errHeldReleaseNotYet):
160 return http.StatusTooEarly
161 case errors.Is(err, release.ErrHeldNotProven):
162 return http.StatusUnprocessableEntity
163 case errors.Is(err, errHeldReleaseUnavailable), errors.Is(err, errHeldReleaseBusy), errors.Is(err, hostwrite.ErrUnavailable):
164 return http.StatusServiceUnavailable
165 case errors.Is(err, release.ErrStaleHeldStage), errors.Is(err, hostbox.ErrChanged), errors.Is(err, errHeldReleaseUnsafe),
166 errors.Is(err, release.ErrHeldNotAuthorized), errors.Is(err, release.ErrManifestDigest), errors.As(err, &refusal):
167 return http.StatusConflict
168 }
169 return http.StatusInternalServerError
170}
171
172// Serve answers one request for an authenticated peer. Its order: the peer,
173// the request's shape, this box's identity, then the action's own reads,
174// every refusal before any effect.
175func (h *movesHeldReleaseRoot) Serve(ctx context.Context, peer StopPeer, q heldReleaseRequest) (heldReleaseResponse, error) {
176 if !h.complete() {
177 return heldReleaseResponse{}, errHeldReleaseUnavailable
178 }
179 if peer.UID == 0 || peer.PID <= 0 || peer.ProcessStart == "" || peer.BootID == "" || peer.ControllerID == "" {
180 return heldReleaseResponse{}, ErrClaimStopIdentity
181 }
182 if err := q.validate(); err != nil {
183 return heldReleaseResponse{}, err
184 }
185 done, err := h.admit(q.Host)
186 if err != nil {
187 return heldReleaseResponse{}, err
188 }
189 defer done()
190 actor, err := h.Actor(ctx)
191 if err != nil {
192 return heldReleaseResponse{}, err
193 }
194 if !actor.Eligible || actor.Quarantined || actor.Box == "" {
195 return heldReleaseResponse{}, hostbox.ErrChanged
196 }
197 if q.Box != actor.Box {
198 return heldReleaseResponse{}, staleHeldStage("it names box %q and this is %q", q.Box, actor.Box)
199 }
200 if !regionname.SameRegion(actor.HomeRegion, q.Region, regionname.PlacementRegion(actor.Region)) {
201 return heldReleaseResponse{}, staleHeldStage("it names region %q and this box is in %q", q.Region, actor.Region)
202 }
203 var out heldReleaseResponse
204 err = withRootReleaseStore(ctx, h.OpenRelease, func() error {
205 var err error
206 if q.Action == heldReleaseStage {
207 out, err = h.stageRelease(ctx, actor, q)
208 } else {
209 out, err = h.publishRelease(ctx, actor, q)
210 }
211 return err
212 })
213 return out, err
214}
215
216// heldReleaseAuthority is root's own read of what a request names.
217type heldReleaseAuthority struct {
218 claim *hostclaim.Claim
219 rec *provision.Record
220 row *registry.Route
221}
222
223// authority re-reads the claim, the record and the registry row, and
224// refuses a request that is not exactly theirs: the holder's incarnation,
225// stamp, session, generation, account and app; the op's id, kind and
226// phase; this box as the move's target (moving, landing) or as the
227// executing source (restoring, reverting), and not declared lost; the hold
228// op the row must be held by (the move id, or the claim's intent.rollback
229// for a restoring op, ruling J3); the record's move and incarnation; and a
230// registry row, when there is one, held by that op for that account and
231// app.
232func (h *movesHeldReleaseRoot) authority(ctx context.Context, actor hostbox.Actor, q heldReleaseRequest) (heldReleaseAuthority, error) {
233 obs, err := h.Claims(ctx, q.Host)
234 if err != nil {
235 return heldReleaseAuthority{}, fmt.Errorf("host claim for %s: %w", q.Host, err)
236 }
237 if obs.Absent || obs.Claim == nil || obs.Claim.Holder == nil || obs.Claim.Host != q.Host {
238 return heldReleaseAuthority{}, staleHeldStage("the host has no claim")
239 }
240 c, hd := obs.Claim, obs.Claim.Holder
241 if hd.Incarnation != q.Incarnation || hd.Stamp != q.Stamp || hd.Session != q.Session || hd.Generation != q.Generation {
242 return heldReleaseAuthority{}, staleHeldStage("the claim's holder is not the one the request observed")
243 }
244 if hd.Account != q.Account || hd.App != q.App {
245 return heldReleaseAuthority{}, staleHeldStage("the claim is not held by %s/%s", q.Account, q.App)
246 }
247 if c.Op.ID != q.MoveID || c.Op.Kind != q.Kind || c.Op.Phase != q.Phase {
248 return heldReleaseAuthority{}, staleHeldStage("the claim is %s %s at %q", c.Op.Kind, c.Op.ID, c.Op.Phase)
249 }
250 if slices.Contains(c.Lost, actor.Box) {
251 return heldReleaseAuthority{}, staleHeldStage("this box is declared lost")
252 }
253 hold := q.MoveID
254 switch c.Op.Kind {
255 case hostclaim.KindMoving, hostclaim.KindLanding:
256 if c.Op.Intent["target"] != actor.Box {
257 return heldReleaseAuthority{}, staleHeldStage("the claim's target is %q", c.Op.Intent["target"])
258 }
259 case hostclaim.KindRestoring, hostclaim.KindReverting:
260 if c.Op.Exec != actor.Box || c.Op.Intent["source"] != actor.Box {
261 return heldReleaseAuthority{}, staleHeldStage("this box is not the move's executing source")
262 }
263 if c.Op.Kind == hostclaim.KindRestoring {
264 if hold = c.Op.Intent["rollback"]; !provision.ValidRollbackID(hold) {
265 return heldReleaseAuthority{}, staleHeldStage("the claim names no rollback")
266 }
267 }
268 default:
269 return heldReleaseAuthority{}, staleHeldStage("the claim's op is %s", c.Op.Kind)
270 }
271 if q.HoldOp != hold {
272 return heldReleaseAuthority{}, staleHeldStage("the row is held by %q, not %q", hold, q.HoldOp)
273 }
274 rec, err := h.Records(ctx, q.Account, q.App, q.Host)
275 if err != nil {
276 return heldReleaseAuthority{}, fmt.Errorf("record for %s: %w", q.Host, err)
277 }
278 if rec == nil || rec.Host != q.Host || rec.Account != q.Account || rec.App != q.App || rec.Move == nil || rec.Move.ID != q.MoveID || rec.Move.Incarnation != hd.Incarnation {
279 return heldReleaseAuthority{}, staleHeldStage("the record is not this move")
280 }
281 inc, err := hostclaim.Incarnation(rec.Account, rec.App, rec.Host, rec.CreatedAt)
282 if err != nil || inc != hd.Incarnation || hd.Key != provision.RecordKey(rec.Account, rec.App, rec.Host) {
283 return heldReleaseAuthority{}, hostbox.ErrChanged
284 }
285 if c.Op.Kind == hostclaim.KindRestoring {
286 if rb := rec.Move.Rollback; rec.Move.State != provision.MoveRollingBack || rb == nil || rb.ID != hold {
287 return heldReleaseAuthority{}, staleHeldStage("the record is not this rollback")
288 }
289 }
290 row, err := h.Route(q.Host)
291 if err != nil {
292 return heldReleaseAuthority{}, fmt.Errorf("registry row for %s: %w", q.Host, err)
293 }
294 if row != nil && (row.HoldOp != hold || release.RouteAccount(*row) != q.Account || row.App != q.App) {
295 return heldReleaseAuthority{}, staleHeldStage("the registry row is %s/%s held by %q", release.RouteAccount(*row), row.App, row.HoldOp)
296 }
297 return heldReleaseAuthority{claim: c, rec: rec, row: row}, nil
298}
299
300// heldReleaseRoute is the held route a request's effect runs for.
301func heldReleaseRoute(q heldReleaseRequest) registry.Route {
302 return registry.Route{Host: q.Host, Kind: "instance", App: q.App, Account: q.Account, HoldOp: q.HoldOp}
303}
304
305// stageRelease is T7 under the claim, for a move's target (moving or
306// landing) or a dirty return's source (restoring): the version must be the
307// source's reported release and, once pinned, the record's pin (for a
308// return, its recorded recovery release, else the pin). It stages and
309// verifies (release.StageHeldVerified), never touches live/<host>, selects
310// nothing and holds no hostguard: the version store is shared and a stage
311// is no host's write.
312func (h *movesHeldReleaseRoot) stageRelease(ctx context.Context, actor hostbox.Actor, q heldReleaseRequest) (heldReleaseResponse, error) {
313 a, err := h.authority(ctx, actor, q)
314 if err != nil {
315 return heldReleaseResponse{}, err
316 }
317 m := a.rec.Move
318 switch q.Kind {
319 case hostclaim.KindMoving, hostclaim.KindLanding:
320 raw, _, err := h.Store.Get(ctx, moves.SourceReportKey(moves.ReportPrefix(q.Account, q.App, q.Host, q.MoveID)))
321 if errors.Is(err, os.ErrNotExist) {
322 return heldReleaseResponse{}, staleHeldStage("the source has not reported its release")
323 }
324 if err != nil {
325 return heldReleaseResponse{}, fmt.Errorf("source report for %s: %w", q.Host, err)
326 }
327 var sr moves.SourceReport
328 if err := json.Unmarshal(raw, &sr); err != nil {
329 return heldReleaseResponse{}, fmt.Errorf("source report for %s: %w", q.Host, err)
330 }
331 if sr.Release == "" || sr.Release != q.Version {
332 return heldReleaseResponse{}, staleHeldStage("the source runs %q", sr.Release)
333 }
334 if m.Inputs != nil && m.Inputs.Release != q.Version {
335 return heldReleaseResponse{}, staleHeldStage("the pinned release is %q", m.Inputs.Release)
336 }
337 case hostclaim.KindRestoring:
338 if want := heldReturnRelease(m); want == "" || want != q.Version {
339 return heldReleaseResponse{}, staleHeldStage("the return's release is %q", want)
340 }
341 default:
342 return heldReleaseResponse{}, staleHeldStage("a %s op stages nothing", q.Kind)
343 }
344 stage := h.stage
345 if stage == nil {
346 pins := h.heldGenerationPins(ctx, q.Version)
347 stage = func(ctx context.Context, s release.Store, pub ed25519.PublicKey, ratchet *release.DigestRatchet, opt string, rt registry.Route, version string, proof *release.ProofGate) (*release.Manifest, error) {
348 return release.StageHeldVerifiedWithPins(ctx, s, pub, ratchet, opt, rt, version, proof, pins)
349 }
350 }
351 if err := withHeldStageLock(h.DataDir, q.Host, func() error {
352 _, err := stage(ctx, h.Store, h.Pub, h.Ratchet(), h.OptDir, heldReleaseRoute(q), q.Version, h.Proof())
353 return err
354 }); err != nil {
355 return heldReleaseResponse{}, err
356 }
357 log.Printf("moves: %q: held release %q staged for %q (no publication)", q.Host, q.Version, q.MoveID)
358 return heldReleaseResponse{Schema: heldReleaseSchema, RequestID: q.RequestID, Outcome: heldReleaseOutcomeStaged}, nil
359}
360
361// heldReturnRelease is a return's release: its recorded recovery release,
362// else the source's pin.
363func heldReturnRelease(m *provision.MoveState) string {
364 if rb := m.Rollback; rb != nil && rb.RecoveryInputs != nil && rb.RecoveryInputs.Release != "" {
365 return rb.RecoveryInputs.Release
366 }
367 if m.Inputs != nil {
368 return m.Inputs.Release
369 }
370 return ""
371}
372
373// publishVersion is the release a publication may name, by the claim's
374// kind and phase, and "" with the reason when the claim grants none: a
375// landing's configured or probing step, on this box as the executor, with
376// the record at that landing phase (the pinned release); a dirty return's
377// configured step (claimed) or its probe (its recovery release, the pin at
378// configured when none is recorded yet). Anything else, a moving op
379// included, publishes nothing: the policy is never widened.
380//
381// A revert's probe and a clean return's probe are refused outright (fix
382// round 2, P2-1, platform-1e's authority review). Neither runs configured,
383// so no render is journalled under a rule root checks, and the journal is
384// agent-written: its absence cannot be what refuses them, since a
385// compromised agent could plant one. They stay refused until those probes
386// journal their own render (RenderHeldUnderBefore at an equivalent point)
387// under a rule root checks.
388func publishVersion(c *hostclaim.Claim, m *provision.MoveState, box string) (string, error) {
389 if c.Op.Exec != box {
390 return "", staleHeldStage("this box is not the executor")
391 }
392 switch c.Op.Kind {
393 case hostclaim.KindLanding:
394 if c.Op.Phase != provision.LandConfigured && c.Op.Phase != provision.LandProbing {
395 return "", staleHeldStage("a landing at %q publishes nothing", c.Op.Phase)
396 }
397 if m.State != provision.MoveLanding || m.LandingPhase != c.Op.Phase {
398 return "", staleHeldStage("the record is %s at %q", m.State, m.LandingPhase)
399 }
400 if m.Inputs == nil || m.Inputs.Release == "" {
401 return "", staleHeldStage("no pinned release")
402 }
403 return m.Inputs.Release, nil
404 case hostclaim.KindRestoring:
405 rb := m.Rollback
406 switch {
407 case c.Op.Phase == hostclaim.PhaseClaimed && rb.SourcePhase == provision.RbSourceConfigured && rb.Dirty:
408 case c.Op.Phase == hostclaim.PhaseProbing && rb.SourcePhase == provision.RbSourceProbing:
409 if !rb.Dirty {
410 return "", staleHeldStage("a clean return's probe journals no render root checks, so root publishes nothing for it")
411 }
412 if rb.RecoveryInputs == nil || rb.RecoveryInputs.Release == "" {
413 return "", staleHeldStage("no recorded recovery release")
414 }
415 default:
416 return "", staleHeldStage("a rollback at %q (source %q) publishes nothing", c.Op.Phase, rb.SourcePhase)
417 }
418 if v := heldReturnRelease(m); v != "" {
419 return v, nil
420 }
421 return "", staleHeldStage("no recorded recovery release")
422 case hostclaim.KindReverting:
423 return "", staleHeldStage("a revert's probe journals no render root checks, so root publishes nothing for it")
424 }
425 return "", staleHeldStage("a %s op publishes nothing", c.Op.Kind)
426}
427
428func (h *movesHeldReleaseRoot) publishRelease(ctx context.Context, actor hostbox.Actor, q heldReleaseRequest) (heldReleaseResponse, error) {
429 a, err := h.authority(ctx, actor, q)
430 if err != nil {
431 return heldReleaseResponse{}, err
432 }
433 want, err := publishVersion(a.claim, a.rec.Move, actor.Box)
434 if err != nil {
435 return heldReleaseResponse{}, err
436 }
437 if q.Version != want {
438 return heldReleaseResponse{}, staleHeldStage("the release to publish is %q", want)
439 }
440 if a.row == nil {
441 return heldReleaseResponse{}, staleHeldStage("the host has no held row")
442 }
443 replayHint := h.replayed(q)
444 if !replayHint {
445 read := h.renderHash
446 if read == nil {
447 read = readHeldRenderHash
448 }
449 rendered, err := read(h.DataDir, q.Host, q.HoldOp)
450 switch {
451 case errors.Is(err, os.ErrNotExist) || (err == nil && rendered == ""):
452 return heldReleaseResponse{}, errHeldReleaseNotYet
453 case err != nil:
454 return heldReleaseResponse{}, err
455 case rendered != q.RenderHash:
456 return heldReleaseResponse{}, staleHeldStage("the journalled render is %s", rendered)
457 }
458
459 }
460 work, err := h.Works(ctx, actor, q.Host, movesActionHeldRelease)
461 if err != nil {
462 return heldReleaseResponse{}, err
463 }
464 sel, err := h.Selected(work)
465 if err != nil {
466 return heldReleaseResponse{}, err
467 }
468 if sel.Host != q.Host || sel.Action != movesActionHeldRelease || sel.Box != actor.Box || sel.Captured.Incarnation != q.Incarnation || sel.Captured.Stamp != q.Stamp ||
469 sel.Session != q.Session || sel.Generation != q.Generation || sel.Kind != q.Kind || sel.Op != q.MoveID || sel.Phase != q.Phase {
470 return heldReleaseResponse{}, hostbox.ErrChanged
471 }
472 checked := heldReleaseCheckedWork{Work: work, check: func() error {
473 fresh, err := h.authority(ctx, actor, q)
474 if err != nil {
475 return err
476 }
477 version, err := publishVersion(fresh.claim, fresh.rec.Move, actor.Box)
478 if err != nil {
479 return err
480 }
481 if version != q.Version || fresh.row == nil {
482 return hostbox.ErrChanged
483 }
484 return nil
485 }}
486 if replayHint {
487 if err := checked.With(ctx, hostwrite.Identity{Release: q.Version}, func() error {
488 if !h.replayed(q) {
489 return hostbox.ErrChanged
490 }
491 return nil
492 }); err != nil {
493 return heldReleaseResponse{}, err
494 }
495 return heldReleaseResponse{Schema: heldReleaseSchema, RequestID: q.RequestID, Outcome: heldReleaseOutcomePublished, Replayed: true}, nil
496 }
497 checked.after = func() error {
498 if h.afterPublish != nil {
499 if err := h.afterPublish(q.Host); err != nil {
500 return err
501 }
502 }
503 if err := h.writeReceipt(q); err != nil {
504 return fmt.Errorf("%w: %v", errHeldReleaseReceipt, err)
505 }
506 return nil
507 }
508 materialize := h.materialize
509 if materialize == nil {
510 pins := h.heldGenerationPins(ctx, q.Version)
511 materialize = func(ctx context.Context, s release.Store, pub ed25519.PublicKey, ratchet *release.DigestRatchet, opt string, rt registry.Route, version string, proof *release.ProofGate, work hostwrite.Work, guard func(func() error) error) (*release.Manifest, error) {
512 return release.MaterializeHeldUnderWithPins(ctx, s, pub, ratchet, opt, rt, version, proof, work, guard, pins)
513 }
514 }
515 lock := func(publish func() error) error { return withHeldStageLock(h.DataDir, q.Host, publish) }
516 if _, err := materialize(ctx, h.Store, h.Pub, h.Ratchet(), h.OptDir, heldReleaseRoute(q), q.Version, h.Proof(), checked, lock); err != nil {
517 return heldReleaseResponse{}, err
518 }
519 log.Printf("moves: %q: held release %q published for %q at %q %q", q.Host, q.Version, q.MoveID, q.Kind, q.Phase)
520 return heldReleaseResponse{Schema: heldReleaseSchema, RequestID: q.RequestID, Outcome: heldReleaseOutcomePublished}, nil
521}
522
523// Serve holds release exclusion until publication returns. Capture before the
524// leaf held-stage lock: retirement may acquire root-host and legacy-host locks.
525// Keep a failed read restrictive only for writes, so an unchanged artifact can
526// still be used and a wholly absent version tree can still be staged.
527func (h *movesHeldReleaseRoot) heldGenerationPins(ctx context.Context, version string) release.PinSource {
528 if h.generationPins == nil {
529 return nil
530 }
531 if _, err := os.Lstat(filepath.Join(h.OptDir, "versions", version)); errors.Is(err, os.ErrNotExist) {
532 return h.generationPins
533 }
534 pins, err := h.generationPins.StagedPins(ctx, h.OptDir)
535 return release.PinSourceFunc(func(context.Context, string) (release.StagedPaths, error) { return pins, err })
536}
537
538type heldReleaseCheckedWork struct {
539 hostwrite.Work
540 check func() error
541 after func() error
542}
543
544func (w heldReleaseCheckedWork) With(ctx context.Context, id hostwrite.Identity, effect func() error) error {
545 return w.Work.With(ctx, id, func() error {
546 if err := w.check(); err != nil {
547 return err
548 }
549 if err := effect(); err != nil {
550 return err
551 }
552 if w.after != nil {
553 return w.after()
554 }
555 return nil
556 })
557}
558
559// heldReleaseReceipt is root's record of a publication it made, one per
560// host (the latest), keyed by the session and the render hash with the
561// identity the publication was for, and the action, kind and phase it was
562// made at (fix round 2, P3-1), so a landing's configured receipt never
563// answers its probing request.
564type heldReleaseReceipt struct {
565 Schema string `json:"schema"`
566 Action string `json:"action"`
567 Kind string `json:"kind"`
568 Phase string `json:"phase"`
569 Host string `json:"host"`
570 Account string `json:"account"`
571 App string `json:"app"`
572 Version string `json:"version"`
573 MoveID string `json:"move_id"`
574 HoldOp string `json:"hold_op"`
575 Incarnation string `json:"incarnation"`
576 Stamp uint64 `json:"stamp"`
577 Session uint64 `json:"session"`
578 Generation string `json:"generation"`
579 RenderHash string `json:"render_hash"`
580 Outcome string `json:"outcome"`
581 At time.Time `json:"at"`
582}
583
584func receiptFor(q heldReleaseRequest) heldReleaseReceipt {
585 return heldReleaseReceipt{Schema: heldReleaseSchema, Action: q.Action, Kind: q.Kind, Phase: q.Phase, Host: q.Host, Account: q.Account, App: q.App, Version: q.Version, MoveID: q.MoveID, HoldOp: q.HoldOp,
586 Incarnation: q.Incarnation, Stamp: q.Stamp, Session: q.Session, Generation: q.Generation, RenderHash: q.RenderHash, Outcome: heldReleaseOutcomePublished}
587}
588
589func (h *movesHeldReleaseRoot) receiptPath(host string) string {
590 return filepath.Join(h.ReceiptDir, host+".json")
591}
592
593// replayed reports whether root's receipt for q's host is for exactly q:
594// the same session and render hash, the same identity, action, kind and
595// phase. Anything else, an unreadable receipt included, is not a replay and
596// runs the whole path.
597func (h *movesHeldReleaseRoot) replayed(q heldReleaseRequest) bool {
598 f, err := safefile.OpenFile(h.receiptPath(q.Host), os.O_RDONLY, 0)
599 if err != nil {
600 return false
601 }
602 defer f.Close()
603 raw, err := io.ReadAll(io.LimitReader(f, heldReleaseRequestMax+1))
604 if err != nil || len(raw) > heldReleaseRequestMax {
605 return false
606 }
607 var got heldReleaseReceipt
608 if err := json.Unmarshal(raw, &got); err != nil {
609 return false
610 }
611 got.At = time.Time{}
612 return got == receiptFor(q)
613}
614
615func (h *movesHeldReleaseRoot) writeReceipt(q heldReleaseRequest) error {
616 r := receiptFor(q)
617 r.At = time.Now().UTC()
618 raw, err := json.Marshal(r)
619 if err != nil {
620 return err
621 }
622 if err := safefile.MkdirAll(h.ReceiptDir, 0o700); err != nil {
623 return err
624 }
625 return safefile.WriteFile(h.receiptPath(q.Host), raw, 0o600)
626}
627