| 1 | package console |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "log" |
| 9 | "net/http" |
| 10 | "os" |
| 11 | "strconv" |
| 12 | "time" |
| 13 | |
| 14 | "github.com/carlosframework/platform/internal/provision" |
| 15 | "github.com/carlosframework/platform/internal/release" |
| 16 | "github.com/carlosframework/platform/internal/runs" |
| 17 | ) |
| 18 | |
| 19 | const runReadFailedMsg = "could not read the run. try again" |
| 20 | |
| 21 | // runCall is one follow route's resolved request. |
| 22 | type runCall struct { |
| 23 | p Principal |
| 24 | actor string |
| 25 | acct *Account |
| 26 | app string |
| 27 | host string |
| 28 | id string |
| 29 | } |
| 30 | |
| 31 | func noRunMsg(host, id string) string { return fmt.Sprintf("%s has no run %s", host, id) } |
| 32 | |
| 33 | // runGate is the four follow routes' shared gate, in the create door's |
| 34 | // order: a person or a run service, a member (the byte-identical 404), the |
| 35 | // host-claim refusal, the runs budget, then a run id of the right shape. A |
| 36 | // service then reaches only a run its own credential started: any other is |
| 37 | // answered exactly as an unknown id is, so a run's existence does not leak. |
| 38 | func (s *Server) runGate(w http.ResponseWriter, r *http.Request) (runCall, bool) { |
| 39 | p, ok := principalFrom(r.Context()) |
| 40 | if !ok || !runCaller(p) { |
| 41 | http.NotFound(w, r) |
| 42 | return runCall{}, false |
| 43 | } |
| 44 | actor, acct, app, ok := s.apiAccountApp(w, r) |
| 45 | if !ok { |
| 46 | return runCall{}, false |
| 47 | } |
| 48 | if s.cfg.HostClaim { |
| 49 | jsonErrorCode(w, http.StatusConflict, runCodeHostClaim, claimRunUnsupportedMsg) |
| 50 | return runCall{}, false |
| 51 | } |
| 52 | if !s.cli.runs.allow(hashToken(bearer(r)), time.Now()) { |
| 53 | jsonErrorCode(w, http.StatusTooManyRequests, runCodeRateLimited, runTooManyMsg) |
| 54 | return runCall{}, false |
| 55 | } |
| 56 | c := runCall{p: p, actor: actor, acct: acct, app: app, host: r.PathValue("host"), id: r.PathValue("id")} |
| 57 | if s.cfg.Releases == nil { |
| 58 | jsonErrorCode(w, http.StatusServiceUnavailable, runCodeUnavailable, runReadFailedMsg) |
| 59 | return runCall{}, false |
| 60 | } |
| 61 | if !provision.ValidHostShape(c.host) || !runs.ValidID(c.id) { |
| 62 | jsonErrorCode(w, http.StatusNotFound, runCodeNotFound, noRunMsg(c.host, c.id)) |
| 63 | return runCall{}, false |
| 64 | } |
| 65 | if p.Kind == PrincipalService { |
| 66 | req, err := runs.ReadRequest(r.Context(), s.cfg.Releases, acct.Sqid, app, c.id) |
| 67 | switch { |
| 68 | case err != nil && !errors.Is(err, os.ErrNotExist): |
| 69 | jsonErrorCode(w, http.StatusBadGateway, runCodeReadFailed, runReadFailedMsg) |
| 70 | return runCall{}, false |
| 71 | case err != nil || !s.runVisibleTo(p, *req): |
| 72 | jsonErrorCode(w, http.StatusNotFound, runCodeNotFound, noRunMsg(c.host, c.id)) |
| 73 | return runCall{}, false |
| 74 | } |
| 75 | } |
| 76 | return c, true |
| 77 | } |
| 78 | |
| 79 | // runOf is the record's run when it is c.id, nil otherwise (the record is |
| 80 | // gone, or names another run). A read failure is an error. |
| 81 | func (s *Server) runOf(ctx context.Context, c runCall) (*provision.RunState, error) { |
| 82 | rec, err := provision.GetRecord(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.host) |
| 83 | if errors.Is(err, os.ErrNotExist) { |
| 84 | return nil, nil |
| 85 | } |
| 86 | if err != nil { |
| 87 | return nil, err |
| 88 | } |
| 89 | if rec.ActiveRun != nil && rec.ActiveRun.ID == c.id { |
| 90 | return rec.ActiveRun, nil |
| 91 | } |
| 92 | return nil, nil |
| 93 | } |
| 94 | |
| 95 | // requestOnHost reports whether the run's objects exist and belong to c.host. |
| 96 | // A run id names one run in the app's subtree; the host in the path must be |
| 97 | // the host the run was admitted for. A read failure is an error. |
| 98 | func (s *Server) requestOnHost(ctx context.Context, c runCall) (bool, error) { |
| 99 | req, err := runs.ReadRequest(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.id) |
| 100 | if errors.Is(err, os.ErrNotExist) { |
| 101 | return false, nil |
| 102 | } |
| 103 | if err != nil { |
| 104 | return false, err |
| 105 | } |
| 106 | return req.Host == c.host, nil |
| 107 | } |
| 108 | |
| 109 | // runWriteCtx is the context for a record write that must run to its end |
| 110 | // even if the caller has gone: a cancelled conditional write may or may not |
| 111 | // have landed, and what follows (an audit row, a delete) depends on which. |
| 112 | func runWriteCtx(ctx context.Context) (context.Context, context.CancelFunc) { |
| 113 | return context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second) |
| 114 | } |
| 115 | |
| 116 | // handleAPIRunGet is GET …/runs/{id}?after=<seq> (run spec §5.6). It reads |
| 117 | // result.json BEFORE it lists chunks: the box writes the result only after |
| 118 | // every chunk is durable, so a page that carries the result also carries |
| 119 | // (or pages on to) every chunk up to its last_seq. Reading them the other |
| 120 | // way round could list the chunks, then see a result written after, and |
| 121 | // tell the caller the run is complete with chunks missing. |
| 122 | func (s *Server) handleAPIRunGet(w http.ResponseWriter, r *http.Request) { |
| 123 | c, ok := s.runGate(w, r) |
| 124 | if !ok { |
| 125 | return |
| 126 | } |
| 127 | ctx := r.Context() |
| 128 | var after int64 |
| 129 | if q := r.URL.Query().Get("after"); q != "" { |
| 130 | var err error |
| 131 | if after, err = strconv.ParseInt(q, 10, 64); err != nil || after < 0 { |
| 132 | jsonErrorCode(w, http.StatusBadRequest, runCodeMalformed, runMalformedMsg) |
| 133 | return |
| 134 | } |
| 135 | } |
| 136 | run, err := s.runOf(ctx, c) |
| 137 | if err != nil { |
| 138 | jsonErrorCode(w, http.StatusBadGateway, runCodeReadFailed, runReadFailedMsg) |
| 139 | return |
| 140 | } |
| 141 | if run == nil { |
| 142 | // The record no longer names the run (cleared, or another run since): |
| 143 | // its objects are what is left of it. |
| 144 | found, err := s.requestOnHost(ctx, c) |
| 145 | switch { |
| 146 | case err != nil: |
| 147 | jsonErrorCode(w, http.StatusBadGateway, runCodeReadFailed, runReadFailedMsg) |
| 148 | return |
| 149 | case !found: |
| 150 | jsonErrorCode(w, http.StatusNotFound, runCodeNotFound, noRunMsg(c.host, c.id)) |
| 151 | return |
| 152 | } |
| 153 | } |
| 154 | res, err := runs.ReadResult(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.id) |
| 155 | if err != nil { |
| 156 | jsonErrorCode(w, http.StatusBadGateway, runCodeReadFailed, runReadFailedMsg) |
| 157 | return |
| 158 | } |
| 159 | keys, err := runs.ListChunks(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.id, after, runs.PageChunks+1) |
| 160 | if err != nil { |
| 161 | jsonErrorCode(w, http.StatusBadGateway, runCodeReadFailed, runReadFailedMsg) |
| 162 | return |
| 163 | } |
| 164 | p := runs.Page{ID: c.id, Result: res, Chunks: []runs.Chunk{}} |
| 165 | if run != nil { |
| 166 | p.State = run.State |
| 167 | } |
| 168 | out := runs.OutPrefix(c.acct.Sqid, c.app, c.id) |
| 169 | size := 0 |
| 170 | for i, k := range keys { |
| 171 | if i == runs.PageChunks { |
| 172 | p.More = true |
| 173 | break |
| 174 | } |
| 175 | seq, stream, ok := runs.ParseChunkKey(k, out) |
| 176 | if !ok { |
| 177 | continue |
| 178 | } |
| 179 | raw, _, err := s.cfg.Releases.Get(ctx, k) |
| 180 | if err != nil { |
| 181 | jsonErrorCode(w, http.StatusBadGateway, runCodeReadFailed, runReadFailedMsg) |
| 182 | return |
| 183 | } |
| 184 | if size+len(raw) > runs.PageBytes && len(p.Chunks) > 0 { |
| 185 | p.More = true |
| 186 | break |
| 187 | } |
| 188 | size += len(raw) |
| 189 | p.Chunks = append(p.Chunks, runs.Chunk{Seq: seq, Stream: stream, Data: raw}) |
| 190 | } |
| 191 | if res != nil { |
| 192 | s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, runEndOutcome(res), res) |
| 193 | } |
| 194 | writeJSON(w, http.StatusOK, p) |
| 195 | } |
| 196 | |
| 197 | // handleAPIRunCancel is POST …/runs/{id}/cancel (run spec §5.7): queued |
| 198 | // becomes cancelled (terminal; the run-end row is written now), picked |
| 199 | // becomes cancel and the serial is bumped so the owning box hears it. |
| 200 | func (s *Server) handleAPIRunCancel(w http.ResponseWriter, r *http.Request) { |
| 201 | c, ok := s.runGate(w, r) |
| 202 | if !ok { |
| 203 | return |
| 204 | } |
| 205 | ctx, cancel := runWriteCtx(r.Context()) |
| 206 | defer cancel() |
| 207 | got, err := provision.SetRecordRun(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.host, provision.RunCancelTransition(c.id)) |
| 208 | if err != nil { |
| 209 | s.runTransitionRefused(w, ctx, c, err) |
| 210 | return |
| 211 | } |
| 212 | switch got.ActiveRun.State { |
| 213 | case provision.RunCancelled: |
| 214 | if err := s.endUnpicked(ctx, c, provision.RunCancelled); err != nil { |
| 215 | log.Printf("console: api run %s/%s/%s %s: end after cancel: %v", c.acct.Sqid, c.app, c.host, c.id, err) |
| 216 | jsonErrorCode(w, http.StatusBadGateway, runCodeRecordFailed, runRecordFailedMsg) |
| 217 | return |
| 218 | } |
| 219 | case provision.RunCancel: |
| 220 | s.bumpSerial(ctx) |
| 221 | } |
| 222 | writeJSON(w, http.StatusOK, runs.StateAnswer{State: got.ActiveRun.State}) |
| 223 | } |
| 224 | |
| 225 | // handleAPIRunExpire is POST …/runs/{id}/expire (run spec §5.2): the CLI's |
| 226 | // "not picked up", asked for sixty seconds after its request. The route |
| 227 | // applies the transition and does not judge the clock; the sweep, which |
| 228 | // expires on its own, decides by the record's pickup Deadline (stamped at |
| 229 | // admission, after the upload), never by request.json's requested_at. A |
| 230 | // pickup that landed first wins. |
| 231 | func (s *Server) handleAPIRunExpire(w http.ResponseWriter, r *http.Request) { |
| 232 | c, ok := s.runGate(w, r) |
| 233 | if !ok { |
| 234 | return |
| 235 | } |
| 236 | ctx, cancel := runWriteCtx(r.Context()) |
| 237 | defer cancel() |
| 238 | got, err := provision.SetRecordRun(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.host, provision.RunExpire(c.id)) |
| 239 | if err != nil { |
| 240 | s.runTransitionRefused(w, ctx, c, err) |
| 241 | return |
| 242 | } |
| 243 | if err := s.endUnpicked(ctx, c, provision.RunExpired); err != nil { |
| 244 | log.Printf("console: api run %s/%s/%s %s: end after expire: %v", c.acct.Sqid, c.app, c.host, c.id, err) |
| 245 | jsonErrorCode(w, http.StatusBadGateway, runCodeRecordFailed, runRecordFailedMsg) |
| 246 | return |
| 247 | } |
| 248 | writeJSON(w, http.StatusOK, runs.StateAnswer{State: got.ActiveRun.State}) |
| 249 | } |
| 250 | |
| 251 | // endUnpicked finishes a run that ended before any box picked it: cancelled |
| 252 | // or expired is terminal, so no box will ever write its result.json. The |
| 253 | // console writes one (create-only, so a repeat changes nothing), so that every |
| 254 | // terminal run has a result a follower can read after the record has moved |
| 255 | // on, and then the run-end row. state is the record's terminal state. |
| 256 | func (s *Server) endUnpicked(ctx context.Context, c runCall, state string) error { |
| 257 | res := runs.Result{Ended: time.Now().UTC()} |
| 258 | if state == provision.RunCancelled { |
| 259 | res.Cancelled = true |
| 260 | } else { |
| 261 | res.Refused = runs.RefusedExpired |
| 262 | } |
| 263 | if err := runs.WriteResult(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.id, res); err != nil { |
| 264 | return err |
| 265 | } |
| 266 | s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, runEndOutcome(&res), &res) |
| 267 | return nil |
| 268 | } |
| 269 | |
| 270 | // runTransitionRefused names why a cancel or expire did not apply, from a |
| 271 | // fresh read of the record. A run the record shows cancelled or expired |
| 272 | // gets its result and run-end row here too (both written once, so a no-op |
| 273 | // when the call that ended it already wrote them): a cancel whose response |
| 274 | // was lost and was retried finds its own terminal state, and these must not |
| 275 | // depend on the first attempt having got that far. |
| 276 | func (s *Server) runTransitionRefused(w http.ResponseWriter, ctx context.Context, c runCall, err error) { |
| 277 | if !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) { |
| 278 | log.Printf("console: api run %s/%s/%s %s: %v", c.acct.Sqid, c.app, c.host, c.id, err) |
| 279 | jsonErrorCode(w, http.StatusBadGateway, runCodeRecordFailed, runRecordFailedMsg) |
| 280 | return |
| 281 | } |
| 282 | run, rerr := s.runOf(ctx, c) |
| 283 | switch { |
| 284 | case rerr != nil: |
| 285 | jsonErrorCode(w, http.StatusBadGateway, runCodeReadFailed, runReadFailedMsg) |
| 286 | case run == nil: |
| 287 | jsonErrorCode(w, http.StatusNotFound, runCodeNotFound, noRunMsg(c.host, c.id)) |
| 288 | case run.Open(): |
| 289 | jsonErrorCode(w, http.StatusConflict, runCodeRunStarted, fmt.Sprintf("run %s has already started", c.id)) |
| 290 | default: |
| 291 | if run.State == provision.RunCancelled || run.State == provision.RunExpired { |
| 292 | if err := s.endUnpicked(ctx, c, run.State); err != nil { |
| 293 | log.Printf("console: api run %s/%s/%s %s: end after retry: %v", c.acct.Sqid, c.app, c.host, c.id, err) |
| 294 | jsonErrorCode(w, http.StatusBadGateway, runCodeRecordFailed, runRecordFailedMsg) |
| 295 | return |
| 296 | } |
| 297 | } |
| 298 | jsonErrorCode(w, http.StatusConflict, runCodeRunEnded, fmt.Sprintf("run %s has already ended", c.id)) |
| 299 | } |
| 300 | } |
| 301 | |
| 302 | // handleAPIRunAck is POST …/runs/{id}/ack (run spec §5.6, §5.8, codex 5): |
| 303 | // refused unless the result exists and last_seq is the result's. Then the |
| 304 | // record's run is cleared and the run's objects are deleted, in that order, |
| 305 | // and only for a run the record does not hold in a live state. |
| 306 | func (s *Server) handleAPIRunAck(w http.ResponseWriter, r *http.Request) { |
| 307 | c, ok := s.runGate(w, r) |
| 308 | if !ok { |
| 309 | return |
| 310 | } |
| 311 | var ack runs.Ack |
| 312 | if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4<<10)).Decode(&ack); err != nil { |
| 313 | jsonErrorCode(w, http.StatusBadRequest, runCodeMalformed, runMalformedMsg) |
| 314 | return |
| 315 | } |
| 316 | ctx, cancel := runWriteCtx(r.Context()) |
| 317 | defer cancel() |
| 318 | found, err := s.requestOnHost(ctx, c) |
| 319 | switch { |
| 320 | case err != nil: |
| 321 | jsonErrorCode(w, http.StatusBadGateway, runCodeReadFailed, runReadFailedMsg) |
| 322 | return |
| 323 | case !found: |
| 324 | jsonErrorCode(w, http.StatusNotFound, runCodeNotFound, noRunMsg(c.host, c.id)) |
| 325 | return |
| 326 | } |
| 327 | res, err := runs.ReadResult(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.id) |
| 328 | switch { |
| 329 | case err != nil: |
| 330 | jsonErrorCode(w, http.StatusBadGateway, runCodeReadFailed, runReadFailedMsg) |
| 331 | return |
| 332 | case res == nil: |
| 333 | jsonErrorCode(w, http.StatusConflict, runCodeRunNotFinished, fmt.Sprintf("run %s has not finished yet", c.id)) |
| 334 | return |
| 335 | case res.LastSeq != ack.LastSeq: |
| 336 | jsonErrorCode(w, http.StatusConflict, runCodeLastSeqMismatch, fmt.Sprintf("run %s ended at chunk %d, not %d", c.id, res.LastSeq, ack.LastSeq)) |
| 337 | return |
| 338 | } |
| 339 | run, err := s.runOf(ctx, c) |
| 340 | switch { |
| 341 | case err != nil: |
| 342 | jsonErrorCode(w, http.StatusBadGateway, runCodeReadFailed, runReadFailedMsg) |
| 343 | return |
| 344 | case run.Open(): |
| 345 | // The box has written the result and not yet moved the run to done. |
| 346 | // Its objects are the box's until it has. |
| 347 | jsonErrorCode(w, http.StatusConflict, runCodeRunNotFinished, fmt.Sprintf("run %s has not finished yet", c.id)) |
| 348 | return |
| 349 | } |
| 350 | s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, runEndOutcome(res), res) |
| 351 | if run != nil { |
| 352 | // Clear the record first. If the CAS fails, or its outcome is unknown, |
| 353 | // the record is asked again before anything is deleted: a run the |
| 354 | // record still names keeps its objects, and the caller may ack again. |
| 355 | _, cerr := provision.SetRecordRun(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.host, provision.RunClear(c.id)) |
| 356 | still, rerr := s.runOf(ctx, c) |
| 357 | if rerr != nil || still != nil { |
| 358 | log.Printf("console: api run %s/%s/%s %s: clear after ack: %v; record read: %v", c.acct.Sqid, c.app, c.host, c.id, cerr, rerr) |
| 359 | jsonErrorCode(w, http.StatusBadGateway, runCodeRecordFailed, runRecordFailedMsg) |
| 360 | return |
| 361 | } |
| 362 | } |
| 363 | if err := runs.DeleteRun(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.id); err != nil { |
| 364 | // The sweep deletes what is left (Task 5); the ack itself stands. |
| 365 | log.Printf("console: api run %s/%s/%s %s: delete after ack: %v", c.acct.Sqid, c.app, c.host, c.id, err) |
| 366 | } |
| 367 | writeJSON(w, http.StatusOK, runs.StateAnswer{State: "acknowledged"}) |
| 368 | } |
| 369 | |
| 370 | // runEndOutcome is the run-end row's outcome, derived from the result alone so |
| 371 | // that whoever writes the row first (a follower's poll, an ack, a cancel, an |
| 372 | // expiry, the sweep) writes the same word: lost, cancelled, expired, or result. |
| 373 | func runEndOutcome(res *runs.Result) string { |
| 374 | switch { |
| 375 | case res.Interrupted == runs.InterruptedLost: |
| 376 | return "lost" |
| 377 | case res.Cancelled: |
| 378 | return provision.RunCancelled |
| 379 | case res.Refused == runs.RefusedExpired: |
| 380 | return provision.RunExpired |
| 381 | } |
| 382 | return "result" |
| 383 | } |
| 384 | |
| 385 | // auditRunEnd writes the run-end row exactly once per run (run spec §6): |
| 386 | // whoever creates runs/<id>/audited first writes it, from the CLI's poll, |
| 387 | // a cancel, an expiry or the sweep. The actor is the run's caller. It runs |
| 388 | // to its end even if the caller has gone, so a cancelled request cannot |
| 389 | // leave the marker without the row's chance. |
| 390 | func (s *Server) auditRunEnd(ctx context.Context, acct *Account, app, host, id, outcome string, res *runs.Result) { |
| 391 | ctx, cancel := runWriteCtx(ctx) |
| 392 | defer cancel() |
| 393 | if _, err := s.cfg.Releases.Put(ctx, runs.AuditedKey(acct.Sqid, app, id), []byte(outcome), ""); err != nil { |
| 394 | if !errors.Is(err, release.ErrCASConflict) { |
| 395 | log.Printf("console: api run %s/%s/%s %s: audit marker: %v", acct.Sqid, app, host, id, err) |
| 396 | } |
| 397 | return // written already, or the bucket refused: either way not twice |
| 398 | } |
| 399 | // The marker is the last object DeleteRun removes, after request.json. A |
| 400 | // marker Put that succeeds for a run whose request.json is gone therefore |
| 401 | // raced a finished ack: the row was written already. Take the marker back |
| 402 | // and write nothing. |
| 403 | by, requester := "", "" |
| 404 | switch req, err := runs.ReadRequest(ctx, s.cfg.Releases, acct.Sqid, app, id); { |
| 405 | case errors.Is(err, os.ErrNotExist): |
| 406 | if derr := s.cfg.Releases.Delete(ctx, runs.AuditedKey(acct.Sqid, app, id)); derr != nil { |
| 407 | log.Printf("console: api run %s/%s/%s %s: remove stray audit marker: %v", acct.Sqid, app, host, id, derr) |
| 408 | } |
| 409 | return |
| 410 | case err == nil: |
| 411 | by, requester = req.By, req.ServiceLine() |
| 412 | } |
| 413 | d := struct { |
| 414 | ID string `json:"id"` |
| 415 | Outcome string `json:"outcome"` |
| 416 | Result *runs.Result `json:"result,omitempty"` |
| 417 | // Requester is a service's run's line, as on the run-start row. |
| 418 | Requester string `json:"requester,omitempty"` |
| 419 | }{id, outcome, res, requester} |
| 420 | raw, _ := json.Marshal(d) |
| 421 | s.auditActivity(acct, app, by, Activity{Action: "run-end", TargetType: "instance", Target: host, Detail: string(raw)}) |
| 422 | } |
| 423 | |
amadan