amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1// This file is apps/<acct>/<app>/schedules.json end to end (spec §4.1): the
2// member-facing validation shared with the console door, so a typo costs a
3// round trip nothing; the CAS-guarded read/modify/write cycle with one
4// retry on a conflicting writer; and schedule-run.json (spec §4.2), whose
5// etag movement alone — never its content — is the "fire it now" signal.
6
7package schedule
8
9import (
10 "context"
11 "crypto/rand"
12 "encoding/hex"
13 "encoding/json"
14 "errors"
15 "fmt"
16 "io"
17 "net/url"
18 "os"
19 "regexp"
20 "sort"
21 "strings"
22 "time"
23
24 "github.com/carlosframework/platform/internal/release"
25)
26
27const MaxSchedulesPerApp = 20
28
29var nameRE = regexp.MustCompile(`^[a-z0-9][a-z0-9-]{0,31}$`)
30
31func ValidName(name string) bool { return nameRE.MatchString(name) }
32
33// ValidPath: absolute, no scheme/host, no "..", short, and rejects whatever
34// url.ParseRequestURI rejects, plus a scheme/host/userinfo of its own. It is
35// NOT the same parser BuildRequest sends through (http.NewRequest /
36// url.Parse): the two differ on "#" — a fragment passes validation here and
37// is silently dropped from the tick, which is harmless since nothing reads
38// it. A path is what the tick POSTs to on the instance's own socket;
39// nothing here can reach another host.
40func ValidPath(p string) error {
41 switch {
42 case p == "" || p[0] != '/':
43 return errors.New("path must start with /")
44 case strings.HasPrefix(p, "//"):
45 return errors.New("path must not start with //")
46 case len(p) > 256:
47 return errors.New("path is longer than 256 bytes")
48 case strings.ContainsAny(p, " \t\r\n"):
49 return errors.New("path must not contain whitespace")
50 }
51 u, err := url.ParseRequestURI(p)
52 if err != nil {
53 return fmt.Errorf("path: %w", err)
54 }
55 if u.Scheme != "" || u.Host != "" || u.User != nil {
56 return errors.New("path must not carry a scheme or host")
57 }
58 for _, seg := range strings.Split(u.Path, "/") {
59 if seg == ".." {
60 return errors.New("path must not contain ..")
61 }
62 }
63 return nil
64}
65
66// Schedule is one declared recurring job, as the member typed it.
67type Schedule struct {
68 Name string `json:"name"`
69 Every string `json:"every,omitempty"`
70 Cron string `json:"cron,omitempty"`
71 Path string `json:"path"`
72}
73
74// Spec is the schedule as the box stores it. Exactly one of every/cron.
75func (s Schedule) Spec() (Spec, error) {
76 switch {
77 case s.Every != "" && s.Cron != "":
78 return Spec{}, errors.New("give --every or --cron, not both")
79 case s.Every != "":
80 d, err := time.ParseDuration(s.Every)
81 if err != nil {
82 return Spec{}, fmt.Errorf("every: %w", err)
83 }
84 if err := ValidEvery(d); err != nil {
85 return Spec{}, err
86 }
87 return Spec{Kind: KindSchedule, Every: d}, nil
88 case s.Cron != "":
89 c, err := ParseCron(s.Cron)
90 if err != nil {
91 return Spec{}, err
92 }
93 return Spec{Kind: KindSchedule, Cron: c}, nil
94 }
95 return Spec{}, errors.New("give --every <duration> or --cron \"<5 fields>\"")
96}
97
98// Validate checks every field: name, path shape, and exactly one of
99// every/cron parses. It does not mutate the receiver — normalising
100// every/cron to their canonical text (so what lands in the bucket is what
101// ParseSpec reads back) is UpsertSchedule's job, done on its own addressable
102// copy just before storing.
103func (s Schedule) Validate() error {
104 _, err := s.validate()
105 return err
106}
107
108// validate is Validate's work, plus the parsed Spec — UpsertSchedule needs
109// that Spec to normalise every/cron text, and calling Validate then Spec
110// separately parsed the every/cron string twice for one call. Kept private
111// so Validate's public signature (and its non-mutating, error-only
112// contract) does not change.
113func (s Schedule) validate() (Spec, error) {
114 if !ValidName(s.Name) {
115 return Spec{}, fmt.Errorf("name %q: want [a-z0-9][a-z0-9-]{0,31}", s.Name)
116 }
117 if err := ValidPath(s.Path); err != nil {
118 return Spec{}, err
119 }
120 sp, err := s.Spec()
121 if err != nil {
122 return Spec{}, err
123 }
124 // A cron expression can parse and still match nothing ("0 0 30 2 *"):
125 // its Next is the zero time, which a box must never store as a due
126 // instant. Refuse it here, where the member can read the answer.
127 if sp.Next(time.Now()).IsZero() {
128 return Spec{}, errors.New("this expression never fires (no matching minute within the next 5 years)")
129 }
130 return sp, nil
131}
132
133// Declaration is apps/<acct>/<app>/schedules.json (spec §4.1).
134type Declaration struct {
135 Schema int `json:"schema"`
136 UpdatedAt string `json:"updated_at"`
137 Schedules []Schedule `json:"schedules"`
138}
139
140func (d Declaration) Find(name string) (Schedule, bool) {
141 for _, s := range d.Schedules {
142 if s.Name == name {
143 return s, true
144 }
145 }
146 return Schedule{}, false
147}
148
149func DeclarationKey(acct, app string) string { return release.AppPrefix(acct, app) + "schedules.json" }
150func RunKey(acct, app string) string { return release.AppPrefix(acct, app) + "schedule-run.json" }
151func StatusPrefix(acct, app string) string { return release.AppPrefix(acct, app) + "schedule-status/" }
152func StatusKey(acct, app, host string) string { return StatusPrefix(acct, app) + host + ".json" }
153
154// ReadDeclaration: absent is the empty declaration with etag "" — legal
155// forever, like restart-gen.
156func ReadDeclaration(ctx context.Context, s release.Store, acct, app string) (Declaration, string, error) {
157 if err := checkIDs(acct, app); err != nil {
158 return Declaration{}, "", err
159 }
160 raw, etag, err := s.Get(ctx, DeclarationKey(acct, app))
161 if errors.Is(err, os.ErrNotExist) {
162 return Declaration{Schema: 1}, "", nil
163 }
164 if err != nil {
165 return Declaration{}, "", err
166 }
167 var d Declaration
168 if err := json.Unmarshal(raw, &d); err != nil {
169 return Declaration{}, "", fmt.Errorf("schedules.json: %w", err)
170 }
171 return d, etag, nil
172}
173
174func checkIDs(acct, app string) error {
175 if !release.ValidAccount(acct) {
176 return fmt.Errorf("schedule: invalid account %q", acct)
177 }
178 if !release.ValidApp(app) {
179 return fmt.Errorf("schedule: invalid app %q", app)
180 }
181 return nil
182}
183
184// modify applies fn under CAS with ONE retry on conflict — two members
185// editing the same app at once should both land, not one of them fail.
186func modify(ctx context.Context, s release.Store, acct, app string, now time.Time, fn func(*Declaration) error) (Declaration, error) {
187 if err := checkIDs(acct, app); err != nil {
188 return Declaration{}, err
189 }
190 var last error
191 for attempt := 0; attempt < 2; attempt++ {
192 d, etag, err := ReadDeclaration(ctx, s, acct, app)
193 if err != nil {
194 return Declaration{}, err
195 }
196 if err := fn(&d); err != nil {
197 return Declaration{}, err
198 }
199 sort.Slice(d.Schedules, func(i, j int) bool { return d.Schedules[i].Name < d.Schedules[j].Name })
200 d.Schema = 1
201 d.UpdatedAt = now.UTC().Format(time.RFC3339)
202 raw, err := json.Marshal(d)
203 if err != nil {
204 return Declaration{}, err
205 }
206 if _, err := s.Put(ctx, DeclarationKey(acct, app), raw, etag); err != nil {
207 if errors.Is(err, release.ErrCASConflict) {
208 last = err
209 continue
210 }
211 return Declaration{}, err
212 }
213 return d, nil
214 }
215 return Declaration{}, fmt.Errorf("schedules.json changed underneath us twice: %w", last)
216}
217
218// UpsertSchedule validates sc, normalises its every/cron text to what
219// ParseSpec reads back, then replaces-or-appends it by name.
220func claimRawUpsertSchedule(ctx context.Context, s release.Store, acct, app string, sc Schedule, now time.Time) (Declaration, error) {
221 sp, err := sc.validate() // one parse of every/cron, not Validate then Spec
222 if err != nil {
223 return Declaration{}, err
224 }
225 if sp.Every > 0 {
226 sc.Every = sp.Every.String()
227 } else {
228 sc.Cron = sp.Cron.String()
229 }
230 return modify(ctx, s, acct, app, now, func(d *Declaration) error {
231 for i := range d.Schedules {
232 if d.Schedules[i].Name == sc.Name {
233 d.Schedules[i] = sc
234 return nil
235 }
236 }
237 if len(d.Schedules) >= MaxSchedulesPerApp {
238 return fmt.Errorf("an app may declare at most %d schedules", MaxSchedulesPerApp)
239 }
240 d.Schedules = append(d.Schedules, sc)
241 return nil
242 })
243}
244
245// RemoveSchedule drops name; present reports whether it was there. Removing
246// an absent name still rewrites updated_at — harmless, and it keeps the
247// door's answer uniform.
248func claimRawRemoveSchedule(ctx context.Context, s release.Store, acct, app, name string, now time.Time) (Declaration, bool, error) {
249 present := false
250 d, err := modify(ctx, s, acct, app, now, func(d *Declaration) error {
251 kept := d.Schedules[:0]
252 for _, sc := range d.Schedules {
253 if sc.Name == name {
254 present = true
255 continue
256 }
257 kept = append(kept, sc)
258 }
259 d.Schedules = kept
260 return nil
261 })
262 return d, present, err
263}
264
265// RunSignal is schedule-run.json: a touch object whose etag movement is the
266// signal, exactly restart-gen's shape and rules (release/restartgen.go).
267type RunSignal struct {
268 Schema int `json:"schema"`
269 Name string `json:"name"`
270 TouchedAt string `json:"touched_at"`
271 Nonce string `json:"nonce"`
272}
273
274// TouchRun bumps the run signal for name. A CAS conflict is success: the
275// signal already fired.
276func claimRawTouchRun(ctx context.Context, s release.Store, acct, app, name string, now time.Time) error {
277 if err := checkIDs(acct, app); err != nil {
278 return err
279 }
280 if !ValidName(name) {
281 return fmt.Errorf("schedule: invalid name %q", name)
282 }
283 key := RunKey(acct, app)
284 _, etag, err := s.Get(ctx, key)
285 if err != nil && !errors.Is(err, os.ErrNotExist) {
286 return err
287 }
288 if errors.Is(err, os.ErrNotExist) {
289 etag = ""
290 }
291 nonce := make([]byte, 16)
292 if _, err := io.ReadFull(rand.Reader, nonce); err != nil {
293 return err
294 }
295 raw, err := json.Marshal(RunSignal{Schema: 1, Name: name, TouchedAt: now.UTC().Format(time.RFC3339), Nonce: hex.EncodeToString(nonce)})
296 if err != nil {
297 return err
298 }
299 if _, err := s.Put(ctx, key, raw, etag); err != nil && !errors.Is(err, release.ErrCASConflict) {
300 return err
301 }
302 return nil
303}
304func RemoveSchedule(ctx context.Context, s release.Store, acct, app, name string, now time.Time) (claimResult0 Declaration, claimResult1 bool, claimResult2 error) {
305 claimResult2 = release.WithAppMutation(ctx, s, acct, app, "schedule.RemoveSchedule", func(claimContext context.Context) error {
306 claimResult0, claimResult1, claimResult2 = claimRawRemoveSchedule(claimContext, s, acct, app, name, now)
307 return claimResult2
308 })
309 return
310}
311func TouchRun(ctx context.Context, s release.Store, acct, app, name string, now time.Time) (claimResult0 error) {
312 claimResult0 = release.WithAppMutation(ctx, s, acct, app, "schedule.TouchRun", func(claimContext context.Context) error {
313 claimResult0 = claimRawTouchRun(claimContext, s, acct, app, name, now)
314 return claimResult0
315 })
316 return
317}
318func UpsertSchedule(ctx context.Context, s release.Store, acct, app string, sc Schedule, now time.Time) (claimResult0 Declaration, claimResult1 error) {
319 claimResult1 = release.WithAppMutation(ctx, s, acct, app, "schedule.UpsertSchedule", func(claimContext context.Context) error {
320 claimResult0, claimResult1 = claimRawUpsertSchedule(claimContext, s, acct, app, sc, now)
321 return claimResult1
322 })
323 return
324}
325