| 1 | package daemon |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "time" |
| 7 | ) |
| 8 | |
| 9 | func (t *Tenant) scheduleViews() { |
| 10 | select { |
| 11 | case t.schedulerWake <- struct{}{}: |
| 12 | default: |
| 13 | } |
| 14 | } |
| 15 | |
| 16 | // The timer follows view deadlines and wakes when writes mark a view dirty. |
| 17 | // Slow maintenance keeps its own cadence; idle tenants do not poll view SQL. |
| 18 | func (t *Tenant) runScheduler() { |
| 19 | next := time.Now().Add(time.Second) |
| 20 | nextMaintenance := next |
| 21 | timer := time.NewTimer(time.Until(next)) |
| 22 | defer timer.Stop() |
| 23 | for { |
| 24 | wake := false |
| 25 | select { |
| 26 | case <-t.workCtx.Done(): |
| 27 | return |
| 28 | case <-t.schedulerWake: |
| 29 | wake = true |
| 30 | case <-timer.C: |
| 31 | } |
| 32 | opCtx, done, err := t.beginOperation(t.workCtx) |
| 33 | if err != nil { |
| 34 | next = time.Now().Add(time.Second) |
| 35 | timer.Reset(time.Second) |
| 36 | continue |
| 37 | } |
| 38 | ctx, cancel := context.WithTimeout(opCtx, 30*time.Second) |
| 39 | now := time.Now() |
| 40 | report := func(name string, err error) { |
| 41 | if err != nil && !errors.Is(err, context.Canceled) { |
| 42 | t.app.telemetry.Logger().Error("tenant maintenance failed" , "tenant" , t.meta.Name, "operation" , name, "error" , err) |
| 43 | } |
| 44 | } |
| 45 | if !wake { |
| 46 | report("records" , t.records.Tick(ctx, now.Unix())) |
| 47 | report("custom views" , t.customViews.tickCustomViews(ctx, now.Unix())) |
| 48 | if !now.Before(nextMaintenance) { |
| 49 | if t.blobs != nil { |
| 50 | report("partial uploads" , t.blobs.CleanupMultipart(ctx)) |
| 51 | } |
| 52 | report("expired uploads" , t.sweepExpiredBlobs(ctx, now.Unix())) |
| 53 | report("retention" , t.sweep(ctx, now.Unix())) |
| 54 | report("restore-state" , t.sweepReplicationState(ctx, now.Unix())) |
| 55 | nextMaintenance = now.Add(time.Minute) |
| 56 | } |
| 57 | next = nextMaintenance |
| 58 | } |
| 59 | due, dueErr := t.records.NextDue(ctx, time.Now().Unix()) |
| 60 | report("view schedule" , dueErr) |
| 61 | if dueErr == nil && due > 0 { |
| 62 | viewAt := time.Unix(due, 0) |
| 63 | if viewAt.Before(next) { |
| 64 | next = viewAt |
| 65 | } |
| 66 | } |
| 67 | cancel() |
| 68 | done() |
| 69 | delay := time.Until(next) |
| 70 | if delay < time.Second { |
| 71 | delay = time.Second |
| 72 | next = time.Now().Add(delay) |
| 73 | } |
| 74 | timer.Reset(delay) |
| 75 | } |
| 76 | } |