carlos / platform Public

Download this file

1	package hostbox
2
3	import (
4	"context"
5	"errors"
6	"sort"
7	"strings"
8	"time"
9
10	"github.com/carlosframework/platform/internal/hostclaim"
11	"github.com/carlosframework/platform/internal/lease"
12	"github.com/carlosframework/platform/internal/park"
13	)
14
15	// RegionalEnrollment must verify current enrollment, cloud identity, region,
16	// bucket access and retirement assignment on every call. Construction-time
17	// configuration is not authority.
18	type RegionalEnrollment interface {
19	VerifyRegional(context.Context, string, string, time.Duration) error
20	}
21
22	// RegionalEvidence returns immutable, durable evidence covering every writer
23	// of residue represented by this exact claim observation. EvidenceAt is the
24	// original evidence time, never a retry time. LegacyIncarnation is required
25	// when the legacy instances/<host>/ namespace contains residue.
26	type RegionalEvidence interface {
27	VerifyRetirement(context.Context, hostclaim.Observed, []lease.ObjectVersion) (legacyIncarnation string, evidenceAt time.Time, err error)
28	}
29
30	// RegionalObligationEvidence binds legacy publication chores before their
31	// first storage effect. It is required only while callers still publish
32	// obligations without the immutable ParkScope.
33	type RegionalObligationEvidence interface {
34	VerifyObligation(context.Context, hostclaim.Observed, hostclaim.Obligation) (hostclaim.ParkScope, error)
35	}
36
37	// RegionalWorker is one pass of the regional retirement worker. Production
38	// wiring must provide cloud-roster verification and a durable evidence reader;
39	// this adapter deliberately has no permissive defaults.
40	type RegionalWorker struct {
41	Regional Regional
42	Enrollment RegionalEnrollment
43	Evidence RegionalEvidence
44	}
45
46	func (w RegionalWorker) validate(ctx context.Context) error {
47	e := w.Regional.Engine
48	if ctx == nil || w.Enrollment == nil || w.Evidence == nil || e.Store == nil || e.Claims == nil || e.Bucket == "" || e.Region == "" {
49	return ErrUnsupported
50	}
51	return w.Enrollment.VerifyRegional(ctx, e.Region, e.Bucket, RequestLimit)
52	}
53
54	// Census returns every host named by either regional data namespace. The
55	// VersionedStore contract makes each call a complete paginated listing.
56	func (w RegionalWorker) Census(ctx context.Context) ([]string, error) {
57	if err := w.validate(ctx); err != nil {
58	return nil, err
59	}
60	hosts := map[string]bool{}
61	for _, prefix := range []string{"instances/", "moves/"} {
62	versions, err := w.Regional.Engine.Store.ListVersions(ctx, w.Regional.Engine.Bucket, prefix)
63	if err != nil {
64	return nil, err
65	}
66	for _, v := range versions {
67	if !strings.HasPrefix(v.Key, prefix) {
68	return nil, ErrUnsupported
69	}
70	tail := strings.TrimPrefix(v.Key, prefix)
71	host, rest, ok := strings.Cut(tail, "/")
72	key, keyErr := hostclaim.Key(host)
73	if !ok || host == "" || rest == "" || v.VersionID == "" || keyErr != nil || key == "" {
74	return nil, ErrUnsupported
75	}
76	hosts[host] = true
77	}
78	}
79	claimKeys, err := w.Regional.Engine.Claims.List(ctx, "control/hosts/")
80	if err != nil {
81	return nil, err
82	}
83	for _, key := range claimKeys {
84	tail, ok := strings.CutPrefix(key, "control/hosts/")
85	if !ok {
86	return nil, ErrUnsupported
87	}
88	// List is recursive; evidence and exec objects are not claim records.
89	if strings.Contains(tail, "/") {
90	continue
91	}
92	if !strings.HasSuffix(tail, ".json") {
93	return nil, ErrUnsupported
94	}
95	host := strings.TrimSuffix(tail, ".json")
96	want, keyErr := hostclaim.Key(host)
97	if keyErr != nil || want != key {
98	return nil, ErrUnsupported
99	}
100	obs, readErr := hostclaim.Read(ctx, w.Regional.Engine.Claims, host)
101	if readErr != nil {
102	return nil, readErr
103	}
104	for _, o := range obs.Claim.Obligations {
105	if w.Regional.Engine.OwnsRegion(o.Region) && o.Owner == "region:"+o.Region {
106	hosts[host] = true
107	break
108	}
109	}
110	}
111	out := make([]string, 0, len(hosts))
112	for host := range hosts {
113	out = append(out, host)
114	}
115	sort.Strings(out)
116	return out, nil
117	}
118
119	// Run performs one bounded pass. Existing chores run in every claim state;
120	// serving/free claims start sweeps; holderless park_sweep:v1 operations admit
121	// residue from another region under the persisted operation. A missing claim is
122	// refused because legacy/backfill disposition is a separate authority.
123	func (w RegionalWorker) Run(ctx context.Context) error {
124	hosts, err := w.Census(ctx)
125	if err != nil {
126	return err
127	}
128	var errs []error
129	for _, host := range hosts {
130	if err := ctx.Err(); err != nil {
131	errs = append(errs, err)
132	break
133	}
134	if err := w.runHost(ctx, host); err != nil {
135	errs = append(errs, err)
136	}
137	}
138	return errors.Join(errs...)
139	}
140
141	// Loop runs a startup pass immediately, then on each serial notification or
142	// one-minute reconciliation tick. Cancelling ctx stops and joins the loop.
143	func (w RegionalWorker) Loop(ctx context.Context, serial <-chan struct{}, report func(error)) error {
144	if ctx == nil {
145	return ErrUnsupported
146	}
147	run := func() {
148	if err := w.Run(ctx); err != nil && report != nil {
149	report(err)
150	}
151	}
152	run()
153	ticker := time.NewTicker(time.Minute)
154	defer ticker.Stop()
155	for {
156	select {
157	case <-ctx.Done():
158	return ctx.Err()
159	case _, ok := <-serial:
160	if !ok {
161	serial = nil
162	continue
163	}
164	run()
165	case <-ticker.C:
166	run()
167	}
168	}
169	}
170
171	func (w RegionalWorker) runHost(ctx context.Context, host string) error {
172	if err := w.Enrollment.VerifyRegional(ctx, w.Regional.Engine.Region, w.Regional.Engine.Bucket, RequestLimit); err != nil {
173	return err
174	}
175	obs, err := hostclaim.Read(ctx, w.Regional.Engine.Claims, host)
176	if err != nil {
177	return err
178	}
179	if obs.Absent || obs.Claim == nil {
180	return ErrUnsupported
181	}
182	pending := false
183	works := append([]hostclaim.Obligation(nil), obs.Claim.Obligations...)
184	for _, selected := range works {
185	obs, err = hostclaim.Read(ctx, w.Regional.Engine.Claims, host)
186	if err != nil {
187	return err
188	}
189	var o hostclaim.Obligation
190	found := false
191	for _, current := range obs.Claim.Obligations {
192	if current.ID == selected.ID && current.Op == selected.Op {
193	o, found = current, true
194	break
195	}
196	}
197	if !found {
198	continue
199	}
200	if o.Owner != "region:"+o.Region || !w.Regional.Engine.OwnsRegion(o.Region) {
201	continue
202	}
203	if o.Kind != "park-replica" && o.Kind != "park-generation" {
204	return ErrUnsupported
205	}
206	if o.Park != nil && o.Park.RequestLimit != RequestLimit {
207	return ErrUnsupported
208	}
209	if o.Park == nil {
210	provider, ok := w.Evidence.(RegionalObligationEvidence)
211	if !ok {
212	return ErrUnsupported
213	}
214	scope, err := provider.VerifyObligation(ctx, obs, o)
215	if err != nil {
216	return err
217	}
218	if scope.EvidenceAt.IsZero() || scope.RequestLimit != RequestLimit {
219	return ErrUnsupported
220	}
221	if err := w.Enrollment.VerifyRegional(ctx, w.Regional.Engine.Region, w.Regional.Engine.Bucket, RequestLimit); err != nil {
222	return err
223	}
224	if err := w.Regional.Engine.Bind(ctx, park.Work{Host: host, ID: o.ID, Op: o.Op}, scope); err != nil {
225	return err
226	}
227	o.Park = &scope
228	}
229	if o.Kind == "park-replica" && o.Park != nil && strings.HasPrefix(o.Park.Location, "moves/"+host+"/") {
230	// park-staging uses the existing park-replica shape and moves scope.
231	}
232	if err := w.Regional.Run(ctx, park.Work{Host: host, ID: o.ID, Op: o.Op}); err != nil {
233	if errors.Is(err, park.ErrPending) {
234	pending = true
235	continue
236	}
237	return err
238	}
239	}
240
241	// Chore execution may have changed or freed the claim. Re-read before a
242	// possible sweep and require fresh enrollment plus evidence for that read.
243	obs, err = hostclaim.Read(ctx, w.Regional.Engine.Claims, host)
244	if err != nil {
245	return err
246	}
247	joining := obs.Claim != nil && obs.Claim.Holder == nil && obs.Claim.Op.Kind == hostclaim.KindParking &&
248	obs.Claim.Op.ID != "" && obs.Claim.Op.Exec == "" && obs.Claim.Op.Intent["park_sweep"] == "v1"
249	if obs.Absent || obs.Claim == nil || (!obs.Claim.Free() && obs.Claim.Op.Kind != hostclaim.KindServing && !joining) {
250	return nil
251	}
252	if pending {
253	return nil
254	}
255	if err := w.Enrollment.VerifyRegional(ctx, w.Regional.Engine.Region, w.Regional.Engine.Bucket, RequestLimit); err != nil {
256	return err
257	}
258	versions, err := w.Regional.Engine.Discover(ctx, host)
259	if err != nil {
260	return err
261	}
262	legacy, evidenceAt, err := w.Evidence.VerifyRetirement(ctx, obs, versions)
263	if err != nil {
264	return err
265	}
266	if evidenceAt.IsZero() {
267	return ErrUnsupported
268	}
269	current, err := hostclaim.Read(ctx, w.Regional.Engine.Claims, host)
270	if err != nil {
271	return err
272	}
273	if current.Absent || current.ETag != obs.ETag {
274	return ErrChanged
275	}
276	if err := w.Enrollment.VerifyRegional(ctx, w.Regional.Engine.Region, w.Regional.Engine.Bucket, RequestLimit); err != nil {
277	return err
278	}
279	verified, err := w.Regional.observe(ctx, host)
280	if err != nil {
281	return err
282	}
283	if verified.ETag != obs.ETag {
284	return ErrChanged
285	}
286	_, err = w.Regional.Engine.SweepObserved(ctx, obs, versions, legacy, evidenceAt, RequestLimit)
287	return err
288	}
289