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 "errors"
6 "fmt"
7 "io"
8 "sync"
9 "time"
10
11 "github.com/carlosframework/platform/internal/activator"
12 "github.com/carlosframework/platform/internal/hostbox"
13 "github.com/carlosframework/platform/internal/hostclaim"
14 "github.com/carlosframework/platform/internal/registry"
15)
16
17type filesSeedIncompleteError struct{ error }
18
19type filesSeedPending struct {
20 exclusion *activator.SeedExclusion
21 wake *ClaimWakeAssembly
22}
23type filesSeedExecutor struct {
24 assembly *ClaimAssembly
25 service *activator.Service
26 registry *registry.Registry
27 source *movesFilesSource
28 mu sync.Mutex
29 running bool
30 pending map[string]filesSeedPending
31}
32
33func (e *filesSeedExecutor) run(ctx context.Context, host string, out io.Writer) error {
34 if e == nil || e.assembly == nil || e.service == nil || e.registry == nil || e.source == nil || out == nil {
35 return hostbox.ErrUnsupported
36 }
37 e.mu.Lock()
38 if e.running {
39 e.mu.Unlock()
40 return activator.ErrFenceHeld
41 }
42 if e.pending == nil {
43 e.pending = map[string]filesSeedPending{}
44 }
45 e.running = true
46 e.mu.Unlock()
47 defer func() { e.mu.Lock(); e.running = false; e.mu.Unlock() }()
48 a := e.assembly
49 if a.Wake == nil || a.Wake.seedFactory == nil {
50 return hostbox.ErrUnsupported
51 }
52 actor, err := a.Identity.Verify(ctx)
53 if err != nil {
54 return err
55 }
56 candidates, err := filesSeedCandidates(ctx, a.Claims, actor.Box, host)
57 if err != nil {
58 return err
59 }
60 var failures []error
61 for _, o := range candidates {
62 if err := ctx.Err(); err != nil {
63 return filesSeedIncompleteError{errors.Join(append(failures, err)...)}
64 }
65 if err := e.host(ctx, o, out); err != nil {
66 failures = append(failures, err)
67 if _, writeErr := fmt.Fprintf(out, "Host %q: failed: %s.\n", o.Host, err); writeErr != nil {
68 return errors.Join(err, writeErr)
69 }
70 }
71 }
72 if err := errors.Join(failures...); err != nil {
73 return filesSeedIncompleteError{err}
74 }
75 return nil
76}
77
78func filesSeedRetryReport(out io.Writer, o hostclaim.Observed) error {
79 if o.Claim == nil || o.Claim.Holder == nil {
80 return nil
81 }
82 r := o.Claim.Holder.WakeRetry
83 if r == nil {
84 r = &hostclaim.WakeRetry{State: "none"}
85 }
86 _, err := fmt.Fprintf(out, "Host %q: wake retry episode %q, state %q, reserved %d of 5, remaining %d, retry after %q.\n", o.Host, r.EpisodeID, r.State, r.AttemptsReserved, 5-r.AttemptsReserved, r.NotBefore.Format(time.RFC3339))
87 return err
88}
89
90func (e *filesSeedExecutor) host(ctx context.Context, o hostclaim.Observed, out io.Writer) (resultErr error) {
91 host := o.Host
92 a := e.assembly
93 actor, err := a.Identity.Verify(ctx)
94 if err != nil {
95 return err
96 }
97 // Scope is decided from the claim before the registry is consulted.
98 if o.Absent || o.Claim == nil {
99 _, err = fmt.Fprintf(out, "Host %q: skipped; no claim.\n", o.Host)
100 return err
101 }
102 if o.Claim.Holder == nil || o.Claim.Op.Kind != hostclaim.KindServing || o.Claim.Op.Exec != actor.Box || o.Claim.Op.ID != "" || o.Claim.Op.Phase != "" {
103 _, err = fmt.Fprintf(out, "Host %q: skipped; claim state %q, executor %q.\n", o.Host, o.Claim.Op.Kind, o.Claim.Op.Exec)
104 return err
105 }
106 row, err := e.registry.Get(o.Host)
107 if err != nil {
108 return err
109 }
110 if row == nil {
111 return hostbox.ErrChanged
112 }
113 if row.Replica {
114 _, err := fmt.Fprintf(out, "Host %q: skipped; replica files are not an authoritative source.\n", o.Host)
115 return err
116 }
117 if row.Seed != "" || row.Backing != "" || row.Unit != "" || row.IsSidecar() || row.Tunnel != "" {
118 return hostbox.ErrUnsupported
119 }
120 opened, err := e.source.Open(ctx, a.Claims, hostbox.RequestStore{Store: a.Data, Check: claimBoundIdentity(a.Identity.Verify, actor)}, actor)
121 if err != nil {
122 return err
123 }
124 files, ok := opened.(*movesFiles)
125 if !ok {
126 return hostbox.ErrUnsupported
127 }
128 if _, err = checkFilesSeedLayout(files, *row, claimEmptyBootstrap(o.Claim.Holder)); err != nil {
129 return err
130 }
131 pending := e.pending[o.Host]
132 retry := o.Claim.Holder.WakeRetry
133 needsSettlement := pending.exclusion != nil || retry != nil && retry.State == "attempting"
134 var token *activator.SeedExclusion
135 release := func() error {
136 if token == nil {
137 return nil
138 }
139 err := token.Release()
140 if err == nil {
141 delete(e.pending, o.Host)
142 token = nil
143 }
144 return err
145 }
146 defer func() {
147 if token != nil && e.pending[o.Host].exclusion == nil {
148 resultErr = errors.Join(resultErr, release())
149 }
150 }()
151 acquire := func() error {
152 var err error
153 token, err = e.service.AcquireSeedExclusion(ctx, o.Host, pending.exclusion)
154 if err != nil {
155 return err
156 }
157 fresh, readErr := hostclaim.Read(ctx, a.Claims, host)
158 if readErr == nil {
159 o = fresh
160 }
161 return readErr
162 }
163 if needsSettlement {
164 if err := acquire(); err != nil {
165 return err
166 }
167 if pending.wake != nil {
168 if err := pending.wake.Close(ctx); err != nil {
169 return err
170 }
171 }
172 settled, settleErr := settleFilesSeedRerun(ctx, *row, files, a.Wake.Retries, o, token)
173 err = settleErr
174 if err != nil {
175 e.pending[o.Host] = filesSeedPending{exclusion: token, wake: pending.wake}
176 return err
177 }
178 delete(e.pending, o.Host)
179 o = settled
180 }
181 selected, err := inspectFilesSeed(ctx, a.Claims, files, o, actor, row)
182 if err != nil {
183 return err
184 }
185 switch selected.State {
186 case "established":
187 if err := release(); err != nil {
188 return err
189 }
190 _, err = fmt.Fprintf(out, "Host %q: an established files manifest already exists; no seed was written.\n", o.Host)
191 return err
192 case "emptyBootstrap":
193 _, err = fmt.Fprintf(out, "Host %q: no files seed needed; the first wake starts empty.\n", o.Host)
194 return err
195 case "replica":
196 _, err = fmt.Fprintf(out, "Host %q: skipped; replica files are not an authoritative source.\n", o.Host)
197 return err
198 case "eligible":
199 default:
200 return hostbox.ErrUnsupported
201 }
202 if token == nil {
203 if err := acquire(); err != nil {
204 return err
205 }
206 selected, err = inspectFilesSeed(ctx, a.Claims, files, o, actor, row)
207 if err != nil {
208 return err
209 }
210 if selected.State != "eligible" {
211 return hostbox.ErrChanged
212 }
213 }
214 if err := filesSeedRetryReport(out, o); err != nil {
215 return err
216 }
217 defer func() {
218 reportCtx, cancel := activator.CleanupContext(ctx, hostbox.RequestLimit)
219 defer cancel()
220 fresh, err := hostclaim.Read(reportCtx, a.Claims, o.Host)
221 if err == nil {
222 err = filesSeedRetryReport(out, fresh)
223 }
224 resultErr = errors.Join(resultErr, err)
225 }()
226 retry = o.Claim.Holder.WakeRetry
227 if retry != nil {
228 if retry.State != "open" || retry.AttemptsReserved >= 5 {
229 return hostclaim.ErrWrongState
230 }
231 delay := time.Until(retry.NotBefore)
232 if delay > 0 {
233 timer := time.NewTimer(delay)
234 defer timer.Stop()
235 select {
236 case <-ctx.Done():
237 return ctx.Err()
238 case <-timer.C:
239 }
240 }
241 }
242 _, op, err := newFilesSeedProvider(ctx, e.source, a.Claims, files.Data, actor, e.registry, selected)
243 if err != nil {
244 return err
245 }
246 runtime, wake, err := a.Wake.seedFactory(ctx, op, token)
247 if err != nil {
248 return err
249 }
250 if _, err = selectFilesSeedInputs(ctx, wake.Inputs, *row, selected); err != nil {
251 return err
252 }
253 admitted, settled, err := prepareFilesSeed(ctx, *row, runtime, o, token)
254 err = errors.Join(err, op.reportSkipped(out, o.Host))
255 if !settled {
256 e.pending[o.Host] = filesSeedPending{exclusion: token, wake: wake}
257 }
258 if admitted.Claim != nil && admitted.Claim.Holder != nil {
259 _, writeErr := fmt.Fprintf(out, "Host %q: admitted files prefix %q; retain this line for interrupted-run cleanup.\n", o.Host, movesFilesPrefix(admitted.Claim.Holder.Location, admitted.Claim.Holder.Generation))
260 err = errors.Join(err, writeErr)
261 }
262 if err != nil {
263 return err
264 }
265 _, err = fmt.Fprintf(out, "Host %q: seeded.\n", o.Host)
266 return err
267}
268
269// close is called only after the serial handler has joined. A failed private
270// runtime stays retained until it joins; shared owners must not close beneath it.
271func (e *filesSeedExecutor) close(ctx context.Context) error {
272 e.mu.Lock()
273 defer e.mu.Unlock()
274 if e.running {
275 return activator.ErrFenceHeld
276 }
277 var result error
278 for _, pending := range e.pending {
279 if pending.wake != nil {
280 result = errors.Join(result, pending.wake.Close(ctx))
281 }
282 }
283 return result
284}
285