| 1 | package console |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "net/http" |
| 6 | "sync" |
| 7 | "time" |
| 8 | |
| 9 | "github.com/carlosframework/platform/internal/hostclaim" |
| 10 | "github.com/carlosframework/platform/internal/hostops" |
| 11 | ) |
| 12 | |
| 13 | type claimCreateAuthority interface { |
| 14 | StartCreate(context.Context, hostclaim.Observed, hostops.Record, string, string) (hostclaim.Observed, error) |
| 15 | RecordEffects(context.Context, hostclaim.Observed, hostops.Record) error |
| 16 | } |
| 17 | |
| 18 | type claimCreateRequest struct { |
| 19 | Record hostops.Record |
| 20 | Observed hostclaim.Observed |
| 21 | Authority claimCreateAuthority |
| 22 | Context context.Context |
| 23 | Close func() |
| 24 | } |
| 25 | |
| 26 | // beginClaimCreate registers through the typed control writer and carries |
| 27 | // the resulting StartHandle in the request context. RecordEffects and the |
| 28 | // private-store issuer take their runner from that handle; nothing derives |
| 29 | // authority from the owner tuple stored in the bucket. |
| 30 | func (s *Server) beginClaimCreate(ctx context.Context, acct *Account, app, host, by string, build func(time.Time, string) (hostops.Record, error)) (*claimCreateRequest, error) { |
| 31 | if !s.cfg.HostClaim || s.cfg.HostClaimOps == nil || s.cfg.ClaimAuthority == nil || s.cfg.ClaimAuthority.Control == nil { |
| 32 | return nil, hostops.ErrContract |
| 33 | } |
| 34 | authority, ok := s.cfg.HostClaimOps.Authority.(claimCreateAuthority) |
| 35 | if !ok { |
| 36 | return nil, hostops.ErrContract |
| 37 | } |
| 38 | before, err := hostclaim.Read(ctx, s.cfg.Releases, host) |
| 39 | if err != nil { |
| 40 | return nil, err |
| 41 | } |
| 42 | // Region HA (spec 2 §7, H27): choose before registering anything. A |
| 43 | // door's earlier choice is kept only if this read is the claim it was |
| 44 | // decided on (census ruling region-ha-2a, item 5). |
| 45 | if ctx, err = s.chooseExecutorFirst(ctx, host, before, build); err != nil { |
| 46 | return nil, err |
| 47 | } |
| 48 | appID, err := s.store.ClaimAppID(acct.ID, app) |
| 49 | if err != nil { |
| 50 | return nil, err |
| 51 | } |
| 52 | admission, err := hostops.ReadAdmission(ctx, s.cfg.Releases, acct.Sqid, app, appID) |
| 53 | if err != nil { |
| 54 | return nil, err |
| 55 | } |
| 56 | id, err := claimEffectID() |
| 57 | if err != nil { |
| 58 | return nil, err |
| 59 | } |
| 60 | controller, err := claimEffectID() |
| 61 | if err != nil { |
| 62 | return nil, err |
| 63 | } |
| 64 | a := s.cfg.ClaimAuthority |
| 65 | owner := hostops.RegistrationOwner{Console: a.ConsoleID, Enrollment: a.Enrollment, Controller: controller} |
| 66 | registration := hostops.Registration{Kind: "create", Host: host, Owner: owner, AppID: appID, Epoch: admission.Value.Epoch, Claim: hostops.SavedObservation(before)} |
| 67 | start, err := a.Control.RegisterStart(ctx, acct.Sqid, app, appID, id, registration, time.Now()) |
| 68 | if err != nil { |
| 69 | return nil, err |
| 70 | } |
| 71 | record, err := build(time.Now().UTC(), id) |
| 72 | if err != nil { |
| 73 | return nil, err |
| 74 | } |
| 75 | inc, err := record.Incarnation() |
| 76 | if err != nil { |
| 77 | return nil, err |
| 78 | } |
| 79 | if err = start.Bind(ctx, record.Holder.CreatedAt, inc, time.Now()); err != nil { |
| 80 | return nil, err |
| 81 | } |
| 82 | requestCtx, cancel := context.WithCancel(ctx) |
| 83 | var wg sync.WaitGroup |
| 84 | wg.Add(1) |
| 85 | go func() { |
| 86 | defer wg.Done() |
| 87 | ticker := time.NewTicker(hostops.RegistrationRenewEvery) |
| 88 | defer ticker.Stop() |
| 89 | for { |
| 90 | select { |
| 91 | case <-requestCtx.Done(): |
| 92 | return |
| 93 | case <-ticker.C: |
| 94 | // An ambiguous renewal stops further work immediately; it |
| 95 | // never extends the acknowledged deadline. |
| 96 | if err := start.Renew(requestCtx, time.Now()); err != nil { |
| 97 | cancel() |
| 98 | return |
| 99 | } |
| 100 | } |
| 101 | } |
| 102 | }() |
| 103 | closeRequest := func() { cancel(); wg.Wait() } |
| 104 | // The start CAS and record effects write claims under this |
| 105 | // registration's lifecycle grant and nothing broader. |
| 106 | startCtx := hostops.WithStartHandle(start.Lifecycle(requestCtx), start) |
| 107 | observed, err := authority.StartCreate(startCtx, before, record, id, by) |
| 108 | if err != nil { |
| 109 | closeRequest() |
| 110 | return nil, err |
| 111 | } |
| 112 | issuer := claimStoreEffects{Runner: start.Runner(time.Now), OperationID: id} |
| 113 | effectCtx := context.WithValue(startCtx, claimEffectContextKey{}, issuer) |
| 114 | return &claimCreateRequest{Record: record, Observed: observed, Authority: authority, Context: effectCtx, Close: closeRequest}, nil |
| 115 | } |
| 116 | |
| 117 | func (s *Server) finishClaimCreate(w http.ResponseWriter, q *claimCreateRequest) bool { |
| 118 | err := q.Authority.RecordEffects(q.Context, q.Observed, q.Record) |
| 119 | if err == nil { |
| 120 | _, err = s.cfg.HostClaimOps.Complete(q.Context, q.Observed.Host, q.Observed.Claim.Op.ID) |
| 121 | } |
| 122 | if err == nil { |
| 123 | return true |
| 124 | } |
| 125 | // The start CAS already accepted this operation. A backend error can |
| 126 | // leave uncertain effects and is observable as pending. |
| 127 | fresh, readErr := hostclaim.Read(context.WithoutCancel(q.Context), s.cfg.Releases, q.Observed.Host) |
| 128 | if readErr == nil && !fresh.Absent && fresh.Claim.Holder != nil && fresh.Claim.Holder.Incarnation == q.Observed.Claim.Holder.Incarnation && fresh.Claim.Op.ID == q.Observed.Claim.Op.ID { |
| 129 | phase := fresh.Claim.Op.Phase |
| 130 | if phase == "" { |
| 131 | phase = fresh.Claim.Op.Kind |
| 132 | } |
| 133 | writeJSON(w, http.StatusAccepted, map[string]string{"operation_id": fresh.Claim.Op.ID, "phase": phase, "host": fresh.Host}) |
| 134 | return false |
| 135 | } |
| 136 | writeClaimResponseError(w, err) |
| 137 | return false |
| 138 | } |
| 139 | |
| 140 | // claimPrivateResponse captures the legacy helper's status without |
| 141 | // committing it to the HTTP connection after lifecycle acceptance. |
| 142 | type claimPrivateResponse struct{ header http.Header } |
| 143 | |
| 144 | func (w *claimPrivateResponse) Header() http.Header { return w.header } |
| 145 | func (w *claimPrivateResponse) Write(body []byte) (int, error) { return len(body), nil } |
| 146 | func (w *claimPrivateResponse) WriteHeader(int) {} |
| 147 | |
amadan