| 1 | package rollout |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "sync" |
| 7 | "time" |
| 8 | |
| 9 | "github.com/carlosframework/platform/internal/activator" |
| 10 | "github.com/carlosframework/platform/internal/registry" |
| 11 | "github.com/carlosframework/platform/internal/release" |
| 12 | ) |
| 13 | |
| 14 | // scheduler is one app's goroutine (spec §6.1). It runs off the agent's |
| 15 | // tick, so a slow rollout never delays the tick or another app. |
| 16 | type scheduler struct { |
| 17 | m *Manager |
| 18 | key appKey |
| 19 | mu sync.Mutex |
| 20 | q *appQueue |
| 21 | kick chan struct{} |
| 22 | } |
| 23 | |
| 24 | func (s *scheduler) poke() { |
| 25 | select { |
| 26 | case s.kick <- struct{}{}: |
| 27 | default: |
| 28 | } |
| 29 | } |
| 30 | |
| 31 | // newest is handed to Swap for its re-checks (spec §6.3). |
| 32 | func (s *scheduler) newest(host string) uint64 { |
| 33 | s.mu.Lock() |
| 34 | defer s.mu.Unlock() |
| 35 | return s.q.newest(host) |
| 36 | } |
| 37 | |
| 38 | func (s *scheduler) loop(ctx context.Context) { |
| 39 | timer := time.NewTimer(time.Hour) |
| 40 | defer timer.Stop() |
| 41 | for { |
| 42 | select { |
| 43 | case <-ctx.Done(): |
| 44 | return |
| 45 | case <-s.kick: |
| 46 | case <-timer.C: |
| 47 | } |
| 48 | if s.m.closing() { |
| 49 | // The agent is stopping (Manager.Close): nothing more is admitted. |
| 50 | return |
| 51 | } |
| 52 | view, pendingConfig := s.view() |
| 53 | for host, rev := range pendingConfig { |
| 54 | if serving, seen := view[host]; seen && serving == "" { |
| 55 | s.m.configSlept(host, rev) |
| 56 | } |
| 57 | } |
| 58 | now := s.m.d.Now() |
| 59 | s.mu.Lock() |
| 60 | ds := s.q.admit(view, now) |
| 61 | next := s.q.nextDue(now) |
| 62 | s.mu.Unlock() |
| 63 | for _, d := range ds { |
| 64 | // Counted before it starts, so Close waits for its done(). One |
| 65 | // admitted as Close began is not run: nothing of it has been |
| 66 | // written yet, and its slot stays taken, which keeps every later |
| 67 | // admission for the host out until the process exits. |
| 68 | if !s.m.enter(true) { |
| 69 | continue |
| 70 | } |
| 71 | go func(d dispatch) { |
| 72 | defer s.m.leave() |
| 73 | s.execute(ctx, d) |
| 74 | }(d) |
| 75 | } |
| 76 | wait := time.Hour |
| 77 | if !next.IsZero() { |
| 78 | wait = max(next.Sub(now), time.Millisecond) |
| 79 | } |
| 80 | timer.Reset(wait) |
| 81 | } |
| 82 | } |
| 83 | |
| 84 | // view asks the activator, with no lock held, what each queued host serves: |
| 85 | // a build when awake, "" when asleep, absent while waking or stopping. It |
| 86 | // also returns each host's pending config revision as it stood BEFORE the |
| 87 | // activator was asked: a host then reported asleep stopped with that file |
| 88 | // already delivered, so its next wake runs it (configSlept). |
| 89 | func (s *scheduler) view() (map[string]string, map[string]string) { |
| 90 | s.mu.Lock() |
| 91 | hosts := make([]string, 0, len(s.q.hosts)) |
| 92 | pendingConfig := map[string]string{} |
| 93 | for name, h := range s.q.hosts { |
| 94 | hosts = append(hosts, name) |
| 95 | if h.wantConfig != "" { |
| 96 | pendingConfig[name] = h.wantConfig |
| 97 | } |
| 98 | } |
| 99 | s.mu.Unlock() |
| 100 | out := make(map[string]string, len(hosts)) |
| 101 | for _, h := range hosts { |
| 102 | if g, ok := s.m.d.Act.Current(h); ok && g.Build != "" { |
| 103 | out[h] = g.Build |
| 104 | } else if !s.m.d.Act.Owns(h) { |
| 105 | out[h] = "" |
| 106 | } |
| 107 | } |
| 108 | return out, pendingConfig |
| 109 | } |
| 110 | |
| 111 | // execute runs one attempt to its outcome. |
| 112 | func (s *scheduler) execute(ctx context.Context, d dispatch) { |
| 113 | m := s.m |
| 114 | rt, err := m.d.Routes.Get(d.ID.Host) |
| 115 | if err != nil || rt == nil { |
| 116 | m.d.Logf("rollout: %s: route not readable for its %s attempt onto %s: %v", d.ID.Host, d.ID.Cause, d.ID.Build, err) |
| 117 | s.done(d, activator.SwapResult{Outcome: activator.SwapAbandoned, Reason: "route not readable"}, "") |
| 118 | return |
| 119 | } |
| 120 | if !Managed(*rt) || release.RouteAccount(*rt) != s.key.acct || rt.App != s.key.app { |
| 121 | // Execution admission (Astra R3 §5 item 4): the attempt was queued |
| 122 | // from the pass's view of the row, and the row this execution runs |
| 123 | // on may have changed since. One that has become a named-unit or |
| 124 | // unit-backed route belongs to root's adoption or to systemd, and |
| 125 | // one that now names another app or account is another scheduler's. |
| 126 | // Either way it is not this attempt's to run. Managed also wants the |
| 127 | // row ready, so one passing through draining, moving or landing is |
| 128 | // not moved now either; the queue keeps it (heldOnBox) and derives |
| 129 | // the attempt again after the deferral. Checked before the |
| 130 | // good-build gate, the staging read and the write-ahead intent, so |
| 131 | // nothing is written and nothing runs: abandoned unexecuted, and |
| 132 | // done() defers the host and re-arms a retry allowance. A row that |
| 133 | // stays unmanaged, or another app's, leaves this queue at the next |
| 134 | // pass (forget). |
| 135 | m.d.Logf("rollout: %s: its %s attempt onto %s is no longer this scheduler's (%s/%s): the route is not a managed route of the app now; not running it", d.ID.Host, d.ID.Cause, d.ID.Build, s.key.acct, s.key.app) |
| 136 | s.done(d, activator.SwapResult{Outcome: activator.SwapAbandoned, Reason: "route no longer managed"}, "") |
| 137 | return |
| 138 | } |
| 139 | if movesBuild(d.ID.Cause) && rt.GoodVersion != "" && d.ID.Build == rt.GoodVersion { |
| 140 | if mark, ok := holdRanFor(m.d.Book, s.key.acct, s.key.app, d.ID.Host); ok && mark.Build != d.ID.Build { |
| 141 | // Ruling 205 at the scheduler (Task 3 fix round 1): a move onto |
| 142 | // the installed good build over another build's hold-ran marker |
| 143 | // is the planner's refused case, for an awake instance, a worker |
| 144 | // or a sidecar member. Root's sibling repoint installs the good |
| 145 | // build too, so only a signed rollback may run it. The queue |
| 146 | // derives such a move as classRecover (the instance has served |
| 147 | // its good build), so the gate is on the cause, not the class: a |
| 148 | // restart or config attempt re-runs the build the instance |
| 149 | // serves and is not gated. Nothing ran: abandoned unexecuted, so |
| 150 | // done() defers the host (abandonDeferral) and the next |
| 151 | // derivation tries again. The pointer pass, due at once on the |
| 152 | // install (observe), clears the marker on a signed rollback. |
| 153 | s.done(d, activator.SwapResult{Outcome: activator.SwapAbandoned, Reason: "a hold build ran here; waiting for a signed rollback"}, "") |
| 154 | return |
| 155 | } |
| 156 | } |
| 157 | a, err := m.resolve(ctx, d, *rt) |
| 158 | if err != nil { |
| 159 | m.stagingRefused(*rt, d.ID.Build, err) |
| 160 | s.mu.Lock() |
| 161 | s.q.deferHost(d.ID.Host, m.d.Now().Add(stagingDeferral)) |
| 162 | s.mu.Unlock() |
| 163 | s.done(d, activator.SwapResult{Outcome: activator.SwapAbandoned, Reason: err.Error()}, "") |
| 164 | return |
| 165 | } |
| 166 | // The failing pointer's PromotedAt, read now, while nothing is at stake: |
| 167 | // done() publishes a failure without a blocking read (spec §8.1). |
| 168 | if d.Class == classAdopt { |
| 169 | d.FailedUnder = m.failedUnder(ctx, *rt, d.ID.Build) |
| 170 | } |
| 171 | // The execution's start, after every read and before the activator runs |
| 172 | // anything: a hold-ran marker this attempt leaves carries it (ruling 205), |
| 173 | // so a rollback signed while it runs is newer than the marker. The book's |
| 174 | // snapshot is taken at the same point, so both describe one start: the |
| 175 | // attempt's good write clears only markers set before it (Astra R3 N6), |
| 176 | // and its failure is stale after a retry since (ruling 191). ExecID names |
| 177 | // this execution for a marker it leaves (plan review finding 4). |
| 178 | d.Started = m.d.Now() |
| 179 | d.BookSeq = m.d.Book.Snapshot() |
| 180 | d.ExecID = m.nextExecID() |
| 181 | if d.Class == classAdopt { |
| 182 | // Write-ahead (Task 5, Astra R1 #5): the adoption's intent is on disk |
| 183 | // before the activator stops or starts anything, so a failure whose |
| 184 | // own save fails is still found by an agent that restarts before a |
| 185 | // save lands (OpenFailureBook). Its mode is the one the activator |
| 186 | // will run (plan review finding 2): Swap runs an overlap on a |
| 187 | // foreign-layout row as a hold, and a reconciled hold holds the |
| 188 | // instance down. A worker's replacement runs its attempt's mode as |
| 189 | // given. No intent, no execution: an attempt whose intent could not |
| 190 | // be written is abandoned unexecuted, and done() defers and retries |
| 191 | // it as any refusal before the exec. Not under scheduler.mu. |
| 192 | mode := a.Mode |
| 193 | if !rt.IsSidecar() { |
| 194 | mode = m.d.Act.EffectiveMode(rt.Host, a.Mode) |
| 195 | } |
| 196 | in := Intent{Account: s.key.acct, App: s.key.app, Build: d.ID.Build, Mode: string(mode), ID: d.ExecID, Start: d.Started} |
| 197 | if err := m.d.Book.Begin(d.ID.Host, in); err != nil { // rolloutsim:M10 |
| 198 | m.d.Logf("rollout: %s: write the intent of its %s attempt onto %s: %v; not running it", d.ID.Host, d.ID.Cause, d.ID.Build, err) |
| 199 | reason := "failure records could not be written" |
| 200 | if errors.Is(err, ErrIntentStands) { |
| 201 | reason = "an earlier execution on this instance has not recorded its outcome" |
| 202 | } |
| 203 | s.done(d, activator.SwapResult{Outcome: activator.SwapAbandoned, Reason: reason}, "") |
| 204 | return |
| 205 | } |
| 206 | } |
| 207 | if rt.IsSidecar() { |
| 208 | // Workers and native sidecar members keep stop-then-start on the |
| 209 | // same entry (spec §7.9). done() runs the cohort-failure transition |
| 210 | // synchronously; only then may a fallback's identity be published |
| 211 | // (spec §8.1), which the activator held back for this call. |
| 212 | s.done(d, s.executeWorker(ctx, *rt, a), a.Mode) |
| 213 | m.d.Act.PublishHeldServing(rt.Host) |
| 214 | return |
| 215 | } |
| 216 | m.emit(*rt, EvSwapping, "info", textSwapping(a.Build, a.Cause)) |
| 217 | s.done(d, m.d.Act.Swap(ctx, a, s.newest), a.Mode) |
| 218 | } |
| 219 | |
| 220 | // movesBuild reports whether an attempt of cause moves the instance onto |
| 221 | // its desired build (a release, install or retry), rather than re-running |
| 222 | // the build it serves (a restart or config attempt). |
| 223 | func movesBuild(c activator.Cause) bool { |
| 224 | return c != activator.CauseRestart && c != activator.CauseConfig |
| 225 | } |
| 226 | |
| 227 | // done records an attempt's outcome and acts on it. mode is the attempt's |
| 228 | // swap mode, "" when no Swap ran. |
| 229 | func (s *scheduler) done(d dispatch, res activator.SwapResult, mode activator.SwapMode) { |
| 230 | m := s.m |
| 231 | if res.Mode != "" { |
| 232 | // The mode the attempt actually ran in: an overlap attempt on a |
| 233 | // foreign-layout row runs as a hold (spec §2.9), and a failed hold |
| 234 | // is held down (§6.4, §8.3). |
| 235 | mode = res.Mode |
| 236 | } |
| 237 | failed := res.Outcome == activator.SwapFailed || res.Outcome == activator.SwapDown |
| 238 | // A failure is published with no blocking read before it (spec §8.1): |
| 239 | // the failing pointer's PromotedAt was read before the attempt ran |
| 240 | // (d.FailedUnder), and the record is written in the same critical |
| 241 | // section that frees the slot and halts the cohort. No admission, no |
| 242 | // newest re-check and no wake plan (which reads the book) sees the slot |
| 243 | // free before the record. The record's build and host are the |
| 244 | // dispatch's, never res.Gen's: a failed hold returns no generation. |
| 245 | var rec *FailureRecord |
| 246 | if failed { |
| 247 | r := FailureRecord{Account: s.key.acct, App: s.key.app, Build: d.ID.Build, Host: d.ID.Host, |
| 248 | Cause: string(d.ID.Cause), Mode: string(mode), Reason: res.Reason, PromotedAt: d.FailedUnder, At: m.d.Now(), |
| 249 | Seq: d.BookSeq, HoldRanAt: d.Started, HoldRanExec: d.ExecID} |
| 250 | if r.PromotedAt.IsZero() { |
| 251 | // No pointer named the build when the attempt started: the |
| 252 | // failure's own time, so only a pointer signed after it clears |
| 253 | // the record. |
| 254 | r.PromotedAt = r.At |
| 255 | } |
| 256 | if res.Outcome == activator.SwapDown && mode == activator.ModeHold { |
| 257 | // A hold build executed against this instance's database: |
| 258 | // nothing falls back on it automatically (spec §6.4, §8.3). |
| 259 | // Held down in the same step that frees the slot, so no wake |
| 260 | // planned in between can boot the failed hold build. |
| 261 | r.HeldDown = []string{d.ID.Host} |
| 262 | } |
| 263 | rec = &r |
| 264 | } |
| 265 | // A failed attempt that carried a config revision leaves it delivered and |
| 266 | // not active (spec §8.1). Its backoff is counted here, before the slot |
| 267 | // frees, and set on the queue in the same step that frees it, so the |
| 268 | // loop cannot re-admit the attempt in between. A failed adoption goes |
| 269 | // through the cohort-failure transition instead, and the next attempt |
| 270 | // for the host is its recovery onto the serving build with the config. |
| 271 | var ( |
| 272 | cfgWait time.Duration |
| 273 | cfgCounted bool |
| 274 | ) |
| 275 | if failed && d.Class == classRecover && d.ID.ConfigRev != "" { |
| 276 | cfgWait, cfgCounted = m.configFailure(d.ID.Host, d.ID.ConfigRev, res.Reason) |
| 277 | } |
| 278 | if res.Outcome == activator.SwapDone { |
| 279 | // The config the attempt started with is active, published under the |
| 280 | // same rule as the serving state (spec §8.1): before the slot frees, |
| 281 | // so a newer config attempt cannot complete first and be overwritten |
| 282 | // here, and with no registry read in front of it, so an unreadable |
| 283 | // row loses nothing. |
| 284 | m.configActivated(d.ID.Host, d.ID.ConfigRev) |
| 285 | // Published BEFORE the slot is freed (spec §8.5): while the attempt |
| 286 | // still holds its host's slot no newer attempt can be admitted and |
| 287 | // complete first, so this write can never land on top of a newer |
| 288 | // one. writeGood also requires the generation to be the one the |
| 289 | // activator serves (acknowledged), and clears only a hold-ran marker |
| 290 | // set before this attempt started (d.BookSeq, Astra R3 N6). |
| 291 | if res.CleanExit { |
| 292 | // A one-shot replacement that exited cleanly (spec §6.5): no |
| 293 | // generation is left for the activator to acknowledge, so the |
| 294 | // acknowledged check is skipped, and only that check (see |
| 295 | // servedOneShot for why that is safe under the held slot). |
| 296 | m.servedOneShot(d.ID.Host, res.Gen, d.BookSeq) |
| 297 | } else { |
| 298 | m.served(d.ID.Host, res.Gen, d.BookSeq) |
| 299 | } |
| 300 | } |
| 301 | now := m.d.Now() |
| 302 | s.mu.Lock() |
| 303 | transition := s.q.finish(d, res.Outcome) |
| 304 | var ( |
| 305 | recorded RecordResult |
| 306 | superseded []string |
| 307 | recErr error |
| 308 | markErr error |
| 309 | ) |
| 310 | // A hold build that ran here and neither passed nor was held down (an |
| 311 | // attempt superseded or stopped after its exec) leaves the instance's |
| 312 | // hold-ran marker (final review wave 2, N2; spec §6.4): the build may |
| 313 | // have migrated the database, so no later wake may boot the good build |
| 314 | // in its place. Only an adoption sets it: a restart or config attempt |
| 315 | // re-runs a build the instance has already served. Set in the step that |
| 316 | // frees the slot, with the failure record when there is one (one save), |
| 317 | // so no wake planned after the slot frees misses it. |
| 318 | heldDown := transition && rec != nil && len(rec.HeldDown) > 0 |
| 319 | mark := mode == activator.ModeHold && d.Class == classAdopt && res.Executed && |
| 320 | res.Outcome != activator.SwapDone && !heldDown |
| 321 | stale := false |
| 322 | if transition && rec != nil { |
| 323 | // The whole cohort-failure transition (spec §8.1 steps 1 and 2): the |
| 324 | // record (in memory at once, on disk as it can) and the halt. None of |
| 325 | // it runs for an attempt that began before a retry of its build |
| 326 | // (PreRetry, ruling 191): the book sets its marker, if any, or one |
| 327 | // in place of its hold-down (r.HoldRanExec, r.HoldRanAt), and |
| 328 | // nothing else. |
| 329 | r := *rec |
| 330 | r.MarkHoldRan = mark |
| 331 | recorded, superseded, recErr = failCohort(m.d.Book, s.q, r) |
| 332 | stale = recorded == PreRetry |
| 333 | } else if mark { |
| 334 | markErr = m.d.Book.MarkHoldRan(d.ID.Host, HoldRan{Account: s.key.acct, App: s.key.app, Build: d.ID.Build, Exec: d.ExecID, At: d.Started}) |
| 335 | } |
| 336 | added := transition && rec != nil && recorded == Recorded |
| 337 | if res.Outcome == activator.SwapAbandoned && !res.Executed { |
| 338 | // It is proven that nothing of the build ran (a refusal before its |
| 339 | // exec, a staging refusal, an unreadable row): a retried host's |
| 340 | // allowance is not spent, and its promised canary still runs after |
| 341 | // the deferral (final review F8). Executed true means the build ran |
| 342 | // or may have (final review wave 2, N4: an uncertain start, a respawn |
| 343 | // whose end the attempt could not learn), and that spends it. |
| 344 | s.q.rearmRetry(d) |
| 345 | } |
| 346 | h := s.q.hosts[d.ID.Host] |
| 347 | if cfgCounted && h != nil && h.wantConfig == d.ID.ConfigRev { |
| 348 | s.q.configFailed(d.ID.Host, now.Add(cfgWait)) |
| 349 | } |
| 350 | if res.Outcome == activator.SwapAbandoned && h != nil && h.buildRev == d.SchedRev && h.notBefore.Before(now.Add(abandonDeferral)) { |
| 351 | // Abandoned while still the host's newest: the same attempt would be |
| 352 | // derived again at once (plan 3's deferral). A staging refusal has |
| 353 | // already deferred the host by as long. |
| 354 | s.q.deferHost(d.ID.Host, now.Add(abandonDeferral)) |
| 355 | } |
| 356 | serving := "" |
| 357 | if h != nil { |
| 358 | serving = h.serving |
| 359 | } |
| 360 | s.mu.Unlock() |
| 361 | // The adoption's intent ends now (Task 5 rule 2), after every book write |
| 362 | // its outcome makes (the record, the hold-ran marker, served's marker |
| 363 | // clear above), never before, and off scheduler.mu. If those writes |
| 364 | // failed to save, this save fails too and the intent stays on disk, where |
| 365 | // a restart finds it. Ended before the poke below, so the next admission |
| 366 | // for the host does not meet it. |
| 367 | var endErr error |
| 368 | if d.Class == classAdopt && d.ExecID != "" { |
| 369 | endErr = m.d.Book.End(d.ID.Host, d.ExecID) |
| 370 | } |
| 371 | if mark || (stale && heldDown) { |
| 372 | // A marker was set: the requested one, or one the book set in place |
| 373 | // of a stale hold-down (Task 4 fix round 1). |
| 374 | m.pointerReadDue(s.key) |
| 375 | } |
| 376 | defer s.poke() |
| 377 | defer m.changed() // runs first: the slot is free and the record is written |
| 378 | if recErr != nil { |
| 379 | m.d.Logf("rollout: write failure record %s/%s %s: %v (held in memory)", rec.Account, rec.App, rec.Build, recErr) |
| 380 | } |
| 381 | if stale { |
| 382 | m.d.Logf("rollout: %s: %s attempt onto %s failed, but it began before a retry of the build: the retry stands", d.ID.Host, d.ID.Cause, d.ID.Build) |
| 383 | } |
| 384 | if markErr != nil { |
| 385 | m.d.Logf("rollout: %s: write hold-ran marker for %s: %v (held in memory)", d.ID.Host, d.ID.Build, markErr) |
| 386 | } |
| 387 | if endErr != nil { |
| 388 | m.d.Logf("rollout: %s: end the intent of its attempt onto %s: %v (it stays on disk until a save lands)", d.ID.Host, d.ID.Build, endErr) |
| 389 | } |
| 390 | if len(superseded) > 0 { |
| 391 | m.d.Logf("rollout: %s/%s build %s failed: superseding attempts past spawn on %v", rec.Account, rec.App, rec.Build, superseded) |
| 392 | } |
| 393 | if failed { |
| 394 | // The raw reason is for the operator; members get the reason word. |
| 395 | m.d.Logf("rollout: %s: %s attempt onto %s %s (%s mode): %s", d.ID.Host, d.ID.Cause, d.ID.Build, res.Outcome, mode, res.Reason) |
| 396 | } |
| 397 | rt := s.route(d.ID.Host) |
| 398 | if added { |
| 399 | m.emit(rt, EvCohortHalted, "warn", textCohortHalted(rec.Build, m.d.Box, reasonWord(rec.Reason))) |
| 400 | } |
| 401 | switch res.Outcome { |
| 402 | case activator.SwapDone: |
| 403 | m.mu.Lock() |
| 404 | if m.retried[d.ID.Host] == d.ID.Build { |
| 405 | delete(m.retried, d.ID.Host) |
| 406 | } |
| 407 | m.mu.Unlock() |
| 408 | m.emit(rt, EvSwapped, "info", textSwapped(d.ID.Build, res.Drain)) |
| 409 | case activator.SwapFailed, activator.SwapDown: |
| 410 | if res.Outcome == activator.SwapDown { |
| 411 | m.emit(rt, EvSwapDown, "error", textSwapDown(d.ID.Build)) |
| 412 | } else { |
| 413 | m.emit(rt, EvSwapFailed, "warn", textSwapFailed(d.ID.Build, serving, reasonWord(res.Reason))) |
| 414 | } |
| 415 | if cfgCounted { |
| 416 | m.emit(rt, EvConfigPending, "warn", textConfigPending(reasonWord(res.Reason), cfgWait)) |
| 417 | } |
| 418 | case activator.SwapAbandoned: |
| 419 | m.d.Logf("rollout: %s: %s attempt onto %s abandoned: %s", d.ID.Host, d.ID.Cause, d.ID.Build, res.Reason) |
| 420 | } |
| 421 | } |
| 422 | |
| 423 | // route is host's row for an event. A row that cannot be read still gets its |
| 424 | // events, addressed by what the scheduler knows: its app and the host. |
| 425 | func (s *scheduler) route(host string) registry.Route { |
| 426 | if rt, err := s.m.d.Routes.Get(host); err == nil && rt != nil { |
| 427 | return *rt |
| 428 | } |
| 429 | return registry.Route{Host: host, Account: s.key.acct, App: s.key.app} |
| 430 | } |
| 431 | |
amadan