carlos / platform Public

Download this file

1	package console
2
3	import (
4	"bytes"
5	"context"
6	"encoding/json"
7	"io"
8	"net/http"
9	"sync"
10	"time"
11
12	"github.com/carlosframework/platform/internal/hostclaim"
13	"github.com/carlosframework/platform/internal/hostops"
14	"github.com/carlosframework/platform/internal/hostwrite"
15	"github.com/carlosframework/platform/internal/release"
16	)
17
18	type WakeRetryResetDoor struct {
19	production bool
20	Claims release.Store
21	Authority func(context.Context) error
22	StopAttempt func(context.Context, hostclaim.Observed, hostclaim.WakeRetryAttempt) error
23	NewID func() string
24	// lifecycle registers the reset with the holder app's admission and
25	// returns ctx carrying that registration's claim grant. The mounted door
26	// always has it (Server.wakeRetryResetDoor); only a direct caller with no
27	// claim runtime runs without it.
28	lifecycle func(context.Context, hostclaim.Observed) (context.Context, func(), error)
29	}
30
31	// wakeRetryResetKind is the admission registration a reset runs under.
32	const wakeRetryResetKind = hostops.RegistrationWakeRetryReset
33
34	type wakeResetRequest struct {
35	Action string `json:"action"`
36	Incarnation string `json:"incarnation"`
37	Episode string `json:"episode"`
38	RequestID string `json:"request_id"`
39	}
40
41	func (d *WakeRetryResetDoor) reset(ctx context.Context, host, actor string, in wakeResetRequest) error {
42	if d == nil || d.Claims == nil || d.Authority == nil || d.StopAttempt == nil || d.NewID == nil {
43	return hostwrite.ErrUnavailable
44	}
45	if err := d.Authority(ctx); err != nil {
46	return err
47	}
48	if actor == "" || in.Action != "reset-wake-retry" || in.Incarnation == "" || in.Episode == "" || in.RequestID == "" {
49	return hostclaim.ErrWrongState
50	}
51	observed, err := hostclaim.Read(ctx, d.Claims, host)
52	if err != nil {
53	return err
54	}
55	if observed.Claim == nil || observed.Claim.Holder == nil || observed.Claim.Holder.Incarnation != in.Incarnation {
56	return hostclaim.ErrWrongState
57	}
58	audit, _ := json.Marshal(struct{ Actor, RequestID string }{actor, in.RequestID})
59	retry := observed.Claim.Holder.WakeRetry
60	if retry == nil || retry.EpisodeID != in.Episode {
61	return hostclaim.ErrWrongState
62	}
63	// The registration comes before any attempt is stopped: a holder app
64	// closed under another exclusive operation refuses here, with no effect.
65	// The stop receives cancellation on renewal loss, and the claim write
66	// carries the lifecycle grant.
67	write := ctx
68	if d.lifecycle != nil {
69	var finish func()
70	if write, finish, err = d.lifecycle(ctx, observed); err != nil {
71	return err
72	}
73	defer finish()
74	ctx = write
75	}
76	if err := ctx.Err(); err != nil {
77	return err
78	}
79	if retry.State == "attempting" {
80	if retry.Attempt == nil {
81	return hostclaim.ErrWrongState
82	}
83	if err = d.StopAttempt(ctx, observed, *retry.Attempt); err != nil {
84	return err
85	}
86	fresh, err := hostclaim.Read(ctx, d.Claims, host)
87	if err != nil {
88	return err
89	}
90	if fresh.ETag != observed.ETag || fresh.Claim == nil || fresh.Claim.Holder == nil || fresh.Claim.Holder.Incarnation != in.Incarnation {
91	return hostclaim.ErrWrongState
92	}
93	r := fresh.Claim.Holder.WakeRetry
94	if r == nil || r.EpisodeID != in.Episode || r.Attempt == nil || *r.Attempt != *retry.Attempt {
95	return hostclaim.ErrWrongState
96	}
97	observed = fresh
98	}
99	if err := d.Authority(ctx); err != nil {
100	return err
101	}
102	_, err = hostclaim.Update(write, d.Claims, observed, hostclaim.ResetWakeRetry(hostclaim.WakeRetryReset{HolderIncarnation: in.Incarnation, ObservedEpisode: in.Episode, NewEpisodeID: d.NewID(), OperatorRequest: string(audit), AttemptSettled: true}))
103	return err
104	}
105
106	func (s *Server) handleHostWakeRetryReset(w http.ResponseWriter, r *http.Request, actor string, raw json.RawMessage) {
107	if len(raw) > 16<<10 {
108	w.WriteHeader(http.StatusBadRequest)
109	return
110	}
111	var in wakeResetRequest
112	decoder := json.NewDecoder(bytes.NewReader(raw))
113	decoder.DisallowUnknownFields()
114	if decoder.Decode(&in) != nil || decoder.Decode(new(any)) != io.EOF {
115	w.WriteHeader(http.StatusBadRequest)
116	return
117	}
118	door := s.wakeRetryResetDoor()
119	if door == nil {
120	w.WriteHeader(http.StatusServiceUnavailable)
121	return
122	}
123	if err := door.reset(r.Context(), r.PathValue("host"), actor, in); err != nil {
124	w.WriteHeader(http.StatusConflict)
125	return
126	}
127	w.Header().Set("Content-Type", "application/json")
128	_ = json.NewEncoder(w).Encode(map[string]bool{"ok": true})
129	}
130
131	// wakeRetryResetDoor is the configured door as the route runs it. The route
132	// answers only with HostClaim on, so the door reads and writes claims through
133	// Config.Releases (the guarded data boundary once ConfigureClaimRuntime ran),
134	// never the handle it was configured with, and every claim write carries the
135	// reset's own registration grant: no ordinary permit and no exemption.
136	func (s *Server) wakeRetryResetDoor() *WakeRetryResetDoor {
137	if s.cfg.WakeRetryReset == nil {
138	return nil
139	}
140	d := *s.cfg.WakeRetryReset
141	d.Claims = s.cfg.Releases
142	d.lifecycle = s.registerWakeRetryReset
143	return &d
144	}
145
146	// registerWakeRetryReset registers the reset with the holder app's canonical
147	// admission: kind reset-wake-retry, the holder's exact incarnation and the
148	// claim observation the door read. RegisterStart refuses a closed admission,
149	// so a foreign exclusive operation stops the reset before StopAttempt. The
150	// grant authorizes claim writes only for this holder app and only while the
151	// registration is live. The request renews it until the stop and claim write
152	// finish, then expires it so the admission sweep can settle it.
153	func (s *Server) registerWakeRetryReset(ctx context.Context, observed hostclaim.Observed) (context.Context, func(), error) {
154	a := s.cfg.ClaimAuthority
155	if !s.claimRuntimeConfigured || a == nil || a.Control == nil {
156	return nil, nil, hostwrite.ErrUnavailable
157	}
158	if observed.Absent || observed.Claim == nil || observed.Claim.Holder == nil {
159	return nil, nil, hostclaim.ErrWrongState
160	}
161	h := observed.Claim.Holder
162	account, err := s.store.AccountBySqid(h.Account)
163	if err != nil {
164	return nil, nil, err
165	}
166	if account == nil {
167	return nil, nil, hostops.ErrIdentity
168	}
169	appID, err := s.store.ClaimAppID(account.ID, h.App)
170	if err != nil {
171	return nil, nil, err
172	}
173	admission, err := hostops.ReadAdmission(ctx, s.cfg.Releases, h.Account, h.App, appID)
174	if err != nil {
175	return nil, nil, err
176	}
177	id, err := claimEffectID()
178	if err != nil {
179	return nil, nil, err
180	}
181	controller, err := claimEffectID()
182	if err != nil {
183	return nil, nil, err
184	}
185	owner := hostops.RegistrationOwner{Console: a.ConsoleID, Enrollment: a.Enrollment, Controller: controller}
186	reg := hostops.Registration{Kind: wakeRetryResetKind, Host: observed.Claim.Host, Owner: owner, AppID: admission.Value.AppID, Epoch: admission.Value.Epoch,
187	Claim: hostops.SavedObservation(observed), Incarnation: h.Incarnation, CreatedAt: h.CreatedAt}
188	registeredAt := time.Now()
189	start, err := a.Control.RegisterStart(ctx, h.Account, h.App, reg.AppID, id, reg, registeredAt)
190	if err != nil {
191	return nil, nil, err
192	}
193	requestCtx, stopRenewal := keepWakeResetRegistration(ctx, registeredAt, hostops.RegistrationTTL-hostops.RegistrationSkew, hostops.RegistrationRenewEvery, start.Renew)
194	finish := func() {
195	stopRenewal()
196	cleanup, done := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
197	defer done()
198	_ = start.FinishWakeRetryReset(cleanup)
199	}
200	return start.Lifecycle(requestCtx), finish, nil
201	}
202
203	func keepWakeResetRegistration(ctx context.Context, registeredAt time.Time, liveFor, every time.Duration, renew func(context.Context, time.Time) error) (context.Context, func()) {
204	requestCtx, cancel := context.WithCancel(ctx)
205	deadline := registeredAt.Add(liveFor)
206	if !time.Now().Before(deadline) {
207	cancel()
208	}
209	watchdog := time.AfterFunc(time.Until(deadline), cancel)
210	var wg sync.WaitGroup
211	wg.Add(1)
212	go func() {
213	defer wg.Done()
214	ticker := time.NewTicker(every)
215	defer ticker.Stop()
216	for {
217	select {
218	case <-requestCtx.Done():
219	return
220	case <-ticker.C:
221	started := time.Now()
222	renewCtx, done := context.WithDeadline(requestCtx, deadline)
223	err := renew(renewCtx, started)
224	done()
225	if err != nil || requestCtx.Err() != nil {
226	cancel()
227	return
228	}
229	deadline = started.Add(liveFor)
230	watchdog.Reset(time.Until(deadline))
231	}
232	}
233	}()
234	return requestCtx, func() {
235	cancel()
236	wg.Wait()
237	watchdog.Stop()
238	}
239	}
240