1package daemon23import (4 "context"5 "errors"6 "time"7)89func (t *Tenant) scheduleViews() {10 select {11 case t.schedulerWake <- struct{}{}:12 default:13 }14}1516// 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.18func (t *Tenant) runScheduler() {19 next := time.Now().Add(time.Second)20 nextMaintenance := next21 timer := time.NewTimer(time.Until(next))22 defer timer.Stop()23 for {24 wake := false25 select {26 case <-t.workCtx.Done():27 return28 case <-t.schedulerWake:29 wake = true30 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 continue37 }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 = nextMaintenance58 }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 = viewAt65 }66 }67 cancel()68 done()69 delay := time.Until(next)70 if delay < time.Second {71 delay = time.Second72 next = time.Now().Add(delay)73 }74 timer.Reset(delay)75 }76}