From c45bf7f8930b3368370777c431089a715eb33023 Mon Sep 17 00:00:00 2001 From: Amp Date: Thu, 6 Aug 2026 08:27:12 +0000 Subject: [PATCH] Add manifest-managed service crons Amp-Thread-ID: https://ampcode.com/threads/T-019fd56a-3372-7029-89df-e069c0fe68e4 Co-authored-by: Arjun Komath --- cli/internal/cli/app.go | 11 +- cli/internal/cli/app_test.go | 34 +- cli/internal/cli/types.go | 1 + cli/internal/manifest/manifest.go | 46 +++ cli/internal/manifest/manifest_test.go | 23 +- .../[serviceId]/configuration/page.tsx | 3 + web/app/api/inngest/route.ts | 4 + web/app/api/projects/[id]/services/route.ts | 16 + web/app/api/services/[id]/logs/route.ts | 4 +- .../service/details/crons-section.tsx | 124 ++++++ web/db/schema.ts | 35 ++ web/db/types.ts | 3 + web/lib/inngest/events/index.ts | 6 +- web/lib/inngest/events/service-cron.ts | 5 + web/lib/inngest/functions/crons.ts | 63 +++ web/lib/inngest/functions/index.ts | 2 + .../functions/service-cron-workflow.ts | 20 + web/lib/public-api-routes.ts | 1 + web/lib/public-api.ts | 154 +++++++- web/lib/service-crons.ts | 362 ++++++++++++++++++ web/lib/victoria-logs.ts | 42 +- web/tests/inngest-route.test.ts | 2 + web/tests/log-routes.test.ts | 12 + web/tests/public-api-configuration.test.ts | 1 + web/tests/public-api-plan.test.ts | 74 ++++ web/tests/public-api-source.test.ts | 51 +++ web/tests/service-crons.test.ts | 213 +++++++++++ web/tests/victoria-logs.test.ts | 60 +++ 28 files changed, 1356 insertions(+), 16 deletions(-) create mode 100644 web/components/service/details/crons-section.tsx create mode 100644 web/lib/inngest/events/service-cron.ts create mode 100644 web/lib/inngest/functions/service-cron-workflow.ts create mode 100644 web/lib/service-crons.ts create mode 100644 web/tests/service-crons.test.ts diff --git a/cli/internal/cli/app.go b/cli/internal/cli/app.go index 990de668..5660d86e 100644 --- a/cli/internal/cli/app.go +++ b/cli/internal/cli/app.go @@ -436,6 +436,7 @@ func (a *App) linkCommand() *cobra.Command { HealthCheck *manifest.HealthCheck `json:"healthCheck"` StartCommand *string `json:"startCommand"` Resources *manifest.Resources `json:"resources"` + Crons []manifest.Cron `json:"crons"` } `json:"current"` Management *serviceManagement `json:"management"` } @@ -467,7 +468,7 @@ func (a *App) linkCommand() *cobra.Command { } } } - m := manifest.Manifest{APIVersion: "v1", Target: &manifest.Target{ServiceID: service.ID}, Service: manifest.Service{Name: service.Name, Source: service.Source, Hostname: cfg.Current.Hostname, Ports: ports, Replicas: cfg.Current.Replicas, Placement: placement, HealthCheck: cfg.Current.HealthCheck, StartCommand: cfg.Current.StartCommand, Resources: cfg.Current.Resources}} + m := manifest.Manifest{APIVersion: "v1", Target: &manifest.Target{ServiceID: service.ID}, Service: manifest.Service{Name: service.Name, Source: service.Source, Hostname: cfg.Current.Hostname, Ports: ports, Replicas: cfg.Current.Replicas, Placement: placement, HealthCheck: cfg.Current.HealthCheck, StartCommand: cfg.Current.StartCommand, Resources: cfg.Current.Resources, Crons: cfg.Current.Crons}} if existing != nil { if existing.Manifest.Linked() && existing.Manifest.Target.ServiceID != service.ID { return fmt.Errorf("manifest is linked to service %s; remove target.serviceId before relinking", existing.Manifest.Target.ServiceID) @@ -516,7 +517,11 @@ func (a *App) applyCommand() *cobra.Command { if placement == nil { return errors.New("service.placement is required") } - body := map[string]any{"name": loaded.Manifest.Service.Name, "source": sourcePatch(loaded.Manifest.Service.Source), "hostname": loaded.Manifest.Service.Hostname, "ports": loaded.Manifest.Service.Ports, "healthCheck": loaded.Manifest.Service.HealthCheck, "startCommand": loaded.Manifest.Service.StartCommand, "resources": loaded.Manifest.Service.Resources} + crons := loaded.Manifest.Service.Crons + if crons == nil { + crons = []manifest.Cron{} + } + body := map[string]any{"name": loaded.Manifest.Service.Name, "source": sourcePatch(loaded.Manifest.Service.Source), "hostname": loaded.Manifest.Service.Hostname, "ports": loaded.Manifest.Service.Ports, "healthCheck": loaded.Manifest.Service.HealthCheck, "startCommand": loaded.Manifest.Service.StartCommand, "resources": loaded.Manifest.Service.Resources, "crons": crons} if placement.Mode == "automatic" { body["placement"] = map[string]any{"mode": "automatic", "replicas": loaded.Manifest.Service.Replicas} } else { @@ -636,7 +641,7 @@ func (a *App) finishLink(path string, existing *manifest.Loaded, service service } return a.printLinked(path, target) } - m := manifest.Manifest{APIVersion: "v1", Target: &manifest.Target{ServiceID: serviceID}, Service: manifest.Service{Name: service.Name, Source: service.Source, Hostname: service.Hostname, Ports: service.Ports, Replicas: service.Replicas, Placement: service.Placement, HealthCheck: service.HealthCheck, StartCommand: service.StartCommand, Resources: service.Resources}} + m := manifest.Manifest{APIVersion: "v1", Target: &manifest.Target{ServiceID: serviceID}, Service: manifest.Service{Name: service.Name, Source: service.Source, Hostname: service.Hostname, Ports: service.Ports, Replicas: service.Replicas, Placement: service.Placement, HealthCheck: service.HealthCheck, StartCommand: service.StartCommand, Resources: service.Resources, Crons: service.Crons}} if err := manifest.Save(path, m); err != nil { return err } diff --git a/cli/internal/cli/app_test.go b/cli/internal/cli/app_test.go index 81cd36d2..84cde7cf 100644 --- a/cli/internal/cli/app_test.go +++ b/cli/internal/cli/app_test.go @@ -335,9 +335,12 @@ func TestApplyExactNestedPatchForSources(t *testing.T) { } source := body["source"].(map[string]any) placement := body["placement"].(map[string]any) - if source["type"] != tc.sourceType || placement["mode"] != "automatic" || placement["replicas"] != float64(2) || len(body) != 8 { + if source["type"] != tc.sourceType || placement["mode"] != "automatic" || placement["replicas"] != float64(2) || len(body) != 9 { t.Fatalf("body=%#v", body) } + if crons, ok := body["crons"].([]any); !ok || len(crons) != 0 { + t.Fatalf("omitted crons must be sent as an empty replacement: %#v", body["crons"]) + } if tc.name == "github_clear_root" { rootDir, present := source["rootDir"] if !present || rootDir != nil { @@ -348,6 +351,35 @@ func TestApplyExactNestedPatchForSources(t *testing.T) { } } +func TestLinkAndApplyRoundTripCrons(t *testing.T) { + d := t.TempDir() + var applied map[string]any + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodGet { + w.Write([]byte(`{"target":{"project":{"slug":"app"},"environment":{"name":"prod"},"service":{"id":"s","name":"web"}},"service":{"id":"s","name":"web","source":{"type":"image","image":"nginx"},"ports":[],"replicas":1,"placement":{"mode":"automatic"},"hostname":"web","healthCheck":null,"startCommand":null,"resources":null,"crons":[{"path":"/jobs/nightly","schedule":"0 5 * * *"}]},"management":{"patchable":true,"blockers":[]}}`)) + return + } + json.NewDecoder(r.Body).Decode(&applied) + w.Write([]byte(`{"action":"noop","currentVersion":"sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","desiredVersion":"sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","changes":[]}`)) + })) + defer s.Close() + writeConfig(t, s.URL) + app, _ := testApp(t, d, s.Client()) + if err := execute(app, "link", "--service", "s"); err != nil { + t.Fatal(err) + } + loaded, err := manifest.Load(d) + if err != nil || !reflect.DeepEqual(loaded.Manifest.Service.Crons, []manifest.Cron{{Path: "/jobs/nightly", Schedule: "0 5 * * *"}}) { + t.Fatalf("crons=%#v err=%v", loaded.Manifest.Service.Crons, err) + } + if err := execute(app, "apply", "--yes"); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(applied["crons"], []any{map[string]any{"path": "/jobs/nightly", "schedule": "0 5 * * *"}}) { + t.Fatalf("applied crons=%#v", applied["crons"]) + } +} + func TestApplyPlacementPayloads(t *testing.T) { for _, tc := range []struct { name, yaml string diff --git a/cli/internal/cli/types.go b/cli/internal/cli/types.go index c3155b1b..8405faa4 100644 --- a/cli/internal/cli/types.go +++ b/cli/internal/cli/types.go @@ -51,6 +51,7 @@ type serviceItem struct { HealthCheck *manifest.HealthCheck `json:"healthCheck"` StartCommand *string `json:"startCommand"` Resources *manifest.Resources `json:"resources"` + Crons []manifest.Cron `json:"crons"` } type targetProject struct { ID string `json:"id"` diff --git a/cli/internal/manifest/manifest.go b/cli/internal/manifest/manifest.go index 876a6238..37aca305 100644 --- a/cli/internal/manifest/manifest.go +++ b/cli/internal/manifest/manifest.go @@ -37,6 +37,11 @@ type Service struct { HealthCheck *HealthCheck `json:"healthCheck" yaml:"healthCheck"` StartCommand *string `json:"startCommand" yaml:"startCommand"` Resources *Resources `json:"resources,omitempty" yaml:"resources,omitempty"` + Crons []Cron `json:"crons,omitempty" yaml:"crons,omitempty"` +} +type Cron struct { + Path string `json:"path" yaml:"path"` + Schedule string `json:"schedule" yaml:"schedule"` } type Placement struct { Mode string `json:"mode" yaml:"mode"` @@ -141,6 +146,10 @@ func ApplyDefaults(m *Manifest) { if m.Service.Ports == nil { m.Service.Ports = []Port{} } + for i := range m.Service.Crons { + m.Service.Crons[i].Path = strings.TrimSpace(m.Service.Crons[i].Path) + m.Service.Crons[i].Schedule = strings.TrimSpace(m.Service.Crons[i].Schedule) + } if m.Service.Replicas == 0 { m.Service.Replicas = 1 } @@ -307,6 +316,43 @@ func Validate(m Manifest) error { return errors.New("service.resources.memoryMb must be between 64 and 65536") } } + seenCronPaths := make(map[string]struct{}, len(m.Service.Crons)) + for i, cron := range m.Service.Crons { + if err := validateCronPath(cron.Path); err != nil { + return fmt.Errorf("service.crons[%d].path: %w", i, err) + } + if _, exists := seenCronPaths[cron.Path]; exists { + return fmt.Errorf("service.crons[%d].path must be unique", i) + } + seenCronPaths[cron.Path] = struct{}{} + if len(cron.Schedule) > 255 || len(strings.Fields(cron.Schedule)) != 5 { + return fmt.Errorf("service.crons[%d].schedule must be a five-field cron expression", i) + } + } + return nil +} + +func validateCronPath(value string) error { + if value == "" || len(value) > 2048 || !strings.HasPrefix(value, "/") || strings.HasPrefix(value, "//") { + return errors.New("must begin with exactly one slash and be at most 2048 characters") + } + decoded, err := url.PathUnescape(value) + if err != nil { + return errors.New("must contain valid URL escapes") + } + if !strings.HasPrefix(decoded, "/") || strings.HasPrefix(decoded, "//") || strings.ContainsAny(decoded, "?#\\") { + return errors.New("cannot contain a host override, query, fragment, or backslash") + } + for _, r := range decoded { + if r < 0x20 || r == 0x7f { + return errors.New("cannot contain control characters") + } + } + for _, segment := range strings.Split(decoded, "/") { + if segment == "." || segment == ".." { + return errors.New("cannot contain '.' or '..' segments") + } + } return nil } func (m Manifest) Linked() bool { diff --git a/cli/internal/manifest/manifest_test.go b/cli/internal/manifest/manifest_test.go index b7c5103a..365c808b 100644 --- a/cli/internal/manifest/manifest_test.go +++ b/cli/internal/manifest/manifest_test.go @@ -12,16 +12,37 @@ func base() Manifest { func TestDefaultsAndRoundTrip(t *testing.T) { m := base() m.Service.Replicas = 0 + m.Service.Crons = []Cron{{Path: " /api/cron/digest ", Schedule: " 0 5 * * * "}} b, e := Marshal(m) if e != nil { t.Fatal(e) } got, e := Parse(b) - if e != nil || got.Service.Replicas != 1 || got.Service.Ports == nil { + if e != nil || got.Service.Replicas != 1 || got.Service.Ports == nil || got.Service.Crons[0] != (Cron{Path: "/api/cron/digest", Schedule: "0 5 * * *"}) { t.Fatalf("got=%#v err=%v", got, e) } } +func TestCronValidation(t *testing.T) { + invalidPaths := []string{"//host/job", "/job?x=1", "/job#part", `/job\\next`, "/a/../b", "/a/%2e%2e/b", "/%2fhost", "/bad%zz", "/line%0Abreak"} + for _, path := range invalidPaths { + m := base() + m.Service.Crons = []Cron{{Path: path, Schedule: "0 5 * * *"}} + if err := Validate(m); err == nil { + t.Fatalf("invalid cron path %q accepted", path) + } + } + m := base() + m.Service.Crons = []Cron{{Path: "/jobs/nightly", Schedule: "0 5 * * *"}, {Path: "/jobs/nightly", Schedule: "0 6 * * *"}} + if err := Validate(m); err == nil || !strings.Contains(err.Error(), "unique") { + t.Fatalf("duplicate path error = %v", err) + } + m.Service.Crons = []Cron{{Path: "/jobs/nightly", Schedule: "0 5 * *"}} + if err := Validate(m); err == nil || !strings.Contains(err.Error(), "five-field") { + t.Fatalf("schedule error = %v", err) + } +} + func TestHostnameValidationAndSlugify(t *testing.T) { for _, value := range []string{"", "Upper", "two words", "-leading", "trailing-", "two--hyphens", strings.Repeat("a", 64)} { m := base() diff --git a/web/app/(dashboard)/dashboard/projects/[slug]/[env]/services/[serviceId]/configuration/page.tsx b/web/app/(dashboard)/dashboard/projects/[slug]/[env]/services/[serviceId]/configuration/page.tsx index 6e1f93db..a97fb142 100644 --- a/web/app/(dashboard)/dashboard/projects/[slug]/[env]/services/[serviceId]/configuration/page.tsx +++ b/web/app/(dashboard)/dashboard/projects/[slug]/[env]/services/[serviceId]/configuration/page.tsx @@ -9,6 +9,7 @@ import { DeleteConfirmationDialog } from "@/components/core/delete-confirmation- import { EditableText } from "@/components/core/editable-text"; import { LocalDate } from "@/components/core/local-date"; import { ConfigSection } from "@/components/service/details/config-section"; +import { CronsSection } from "@/components/service/details/crons-section"; import { HealthCheckSection } from "@/components/service/details/health-check-section"; import { NetworkingSection } from "@/components/service/details/networking-section"; import { ReplicasSection } from "@/components/service/details/replicas-section"; @@ -119,6 +120,8 @@ export default function ConfigurationPage() { + +
diff --git a/web/app/api/inngest/route.ts b/web/app/api/inngest/route.ts index 99212b54..3f82d6a0 100644 --- a/web/app/api/inngest/route.ts +++ b/web/app/api/inngest/route.ts @@ -24,6 +24,8 @@ import { scheduledDeploymentsCheck, serviceDeletionWorkflow, serviceCommandRetention, + serviceCronDispatcher, + serviceCronWorkflow, serviceRestoreWorkflow, staleItemsCleanup, staleServerCheck, @@ -58,5 +60,7 @@ export const { GET, POST, PUT } = serve({ expiredDeletedServicesPurge, notificationDelivery, notificationRetention, + serviceCronDispatcher, + serviceCronWorkflow, ], }); diff --git a/web/app/api/projects/[id]/services/route.ts b/web/app/api/projects/[id]/services/route.ts index cb44107a..9ad6b75f 100644 --- a/web/app/api/projects/[id]/services/route.ts +++ b/web/app/api/projects/[id]/services/route.ts @@ -12,6 +12,7 @@ import { servers, servicePorts, serviceReplicas, + serviceCrons, serviceRevisions, services, serviceVolumes, @@ -161,6 +162,20 @@ export async function GET( : and(eq(services.projectId, projectId), isNull(services.deletedAt)), ) .orderBy(services.createdAt); + const cronRows = + servicesList.length > 0 + ? await db + .select() + .from(serviceCrons) + .where( + inArray( + serviceCrons.serviceId, + servicesList.map((service) => service.id), + ), + ) + .orderBy(serviceCrons.path) + : []; + const cronsByService = Map.groupBy(cronRows, (cron) => cron.serviceId); const result = await Promise.all( servicesList.map(async (service) => { @@ -401,6 +416,7 @@ export async function GET( return { ...service, ports, + crons: cronsByService.get(service.id) ?? [], configuredReplicas: replicas, deployments: deploymentsWithDetails, secrets: serviceSecrets, diff --git a/web/app/api/services/[id]/logs/route.ts b/web/app/api/services/[id]/logs/route.ts index 49bcd40b..f4d3f314 100644 --- a/web/app/api/services/[id]/logs/route.ts +++ b/web/app/api/services/[id]/logs/route.ts @@ -34,7 +34,9 @@ export async function GET( const serverId = url.searchParams.get("serverId") || undefined; const logTypeParam = url.searchParams.get("type"); const logType = - logTypeParam === "container" || logTypeParam === "http" + logTypeParam === "container" || + logTypeParam === "http" || + logTypeParam === "cron" ? (logTypeParam as LogType) : undefined; diff --git a/web/components/service/details/crons-section.tsx b/web/components/service/details/crons-section.tsx new file mode 100644 index 00000000..3d7a34e7 --- /dev/null +++ b/web/components/service/details/crons-section.tsx @@ -0,0 +1,124 @@ +"use client"; + +import cronstrue from "cronstrue"; +import { Ban, CheckCircle2, XCircle } from "lucide-react"; +import { memo } from "react"; +import { LocalDate } from "@/components/core/local-date"; +import { ConfigSection } from "@/components/service/details/config-section"; +import { StatusBadge } from "@/components/ui/status-badge"; +import type { ServiceCron, ServiceWithDetails as Service } from "@/db/types"; + +const STATUS_CONFIG = { + succeeded: { + icon: CheckCircle2, + label: "Succeeded", + className: "text-green-500", + }, + failed: { icon: XCircle, label: "Failed", className: "text-red-500" }, + skipped: { icon: Ban, label: "Skipped", className: "text-orange-500" }, +} as const; + +function describeSchedule(schedule: string) { + try { + return cronstrue.toString(schedule, { verbose: true }); + } catch { + return "Invalid cron expression"; + } +} + +function CronStatus({ + status, +}: { + status: NonNullable; +}) { + const config = STATUS_CONFIG[status]; + return ( + + ); +} + +function formatDuration(durationMs: number) { + return durationMs < 1000 + ? `${durationMs}ms` + : `${(durationMs / 1000).toFixed(2)}s`; +} + +export const CronsSection = memo(function CronsSection({ + service, +}: { + service: Service; +}) { + const crons = service.crons ?? []; + + return ( + 0 ? `${crons.length}` : "None"} + summaryMuted={crons.length === 0} + > +
+

+ Cron jobs are managed in{" "} + techulus.yml. Schedules run in UTC + and are read-only here. +

+ + {crons.length === 0 ? ( +

+ No cron jobs configured. +

+ ) : ( +
+ {crons.map((cron) => ( +
+
+

+ {cron.path} +

+

{cron.schedule}

+

+ {describeSchedule(cron.schedule)} (UTC) +

+
+ + {cron.lastStatus ? ( +
+
+ + + + +
+
+ HTTP status: {cron.lastStatusCode ?? "—"} + + Duration:{" "} + {cron.lastDurationMs == null + ? "—" + : formatDuration(cron.lastDurationMs)} + +
+ {cron.lastError && ( +

+ {cron.lastError} +

+ )} +
+ ) : ( +

Never run.

+ )} +
+ ))} +
+ )} +
+ + ); +}); diff --git a/web/db/schema.ts b/web/db/schema.ts index 82f23b87..5a80770d 100644 --- a/web/db/schema.ts +++ b/web/db/schema.ts @@ -602,6 +602,41 @@ export const services = pgTable( ], ); +export const serviceCrons = pgTable( + "service_crons", + { + id: text("id").primaryKey(), + serviceId: text("service_id") + .notNull() + .references(() => services.id, { onDelete: "cascade" }), + path: text("path").notNull(), + schedule: text("schedule").notNull(), + nextScheduledFor: timestamp("next_scheduled_for", { + withTimezone: true, + }).notNull(), + lastScheduledFor: timestamp("last_scheduled_for", { withTimezone: true }), + lastAttemptedFor: timestamp("last_attempted_for", { withTimezone: true }), + lastStartedAt: timestamp("last_started_at", { withTimezone: true }), + lastFinishedAt: timestamp("last_finished_at", { withTimezone: true }), + lastStatus: text("last_status", { + enum: ["succeeded", "failed", "skipped"], + }), + lastStatusCode: integer("last_status_code"), + lastDurationMs: integer("last_duration_ms"), + lastError: text("last_error"), + createdAt: timestamp("created_at", { withTimezone: true }) + .defaultNow() + .notNull(), + }, + (table) => [ + uniqueIndex("service_crons_service_path_unique_idx").on( + table.serviceId, + table.path, + ), + index("service_crons_due_scan_idx").on(table.nextScheduledFor), + ], +); + export const serviceReplicas = pgTable( "service_replicas", { diff --git a/web/db/types.ts b/web/db/types.ts index 875fedf4..95849e6d 100644 --- a/web/db/types.ts +++ b/web/db/types.ts @@ -13,6 +13,7 @@ import type { servicePorts, serviceReplicas, serviceCommands, + serviceCrons, services, serviceVolumes, user, @@ -27,6 +28,7 @@ export type ServicePort = typeof servicePorts.$inferSelect; export type ServiceVolume = typeof serviceVolumes.$inferSelect; export type ServiceReplica = typeof serviceReplicas.$inferSelect; export type ServiceCommand = typeof serviceCommands.$inferSelect; +export type ServiceCron = typeof serviceCrons.$inferSelect; export type Secret = typeof secrets.$inferSelect; export type Deployment = typeof deployments.$inferSelect; export type DeploymentPort = typeof deploymentPorts.$inferSelect; @@ -47,6 +49,7 @@ export type ServiceWithDetails = Service & { activeConfig?: DeployedConfig | null; currentSource: SourceConfig; ports: ServicePort[]; + crons: ServiceCron[]; configuredReplicas: Array< ServiceReplica & { serverName: string; serverIsProxy: boolean } >; diff --git a/web/lib/inngest/events/index.ts b/web/lib/inngest/events/index.ts index 0bf59f8d..4e284e71 100644 --- a/web/lib/inngest/events/index.ts +++ b/web/lib/inngest/events/index.ts @@ -8,6 +8,7 @@ export type { ResourceEvents } from "./resource"; export type { RestoreEvents } from "./restore"; export type { RolloutEvents } from "./rollout"; export type { ServiceDeletionEvents } from "./service-deletion"; +export type { ServiceCronEvents } from "./service-cron"; import type { BackupEvents } from "./backup"; import type { BuildEvents } from "./build"; @@ -17,6 +18,7 @@ import type { ResourceEvents } from "./resource"; import type { RestoreEvents } from "./restore"; import type { RolloutEvents } from "./rollout"; import type { ServiceDeletionEvents } from "./service-deletion"; +import type { ServiceCronEvents } from "./service-cron"; export type Events = RolloutEvents & MigrationEvents & @@ -25,7 +27,8 @@ export type Events = RolloutEvents & BuildEvents & ServiceDeletionEvents & ResourceEvents & - NotificationEvents; + NotificationEvents & + ServiceCronEvents; type EventName = keyof Events & string; type EventData = Events[TName]["data"]; @@ -61,4 +64,5 @@ export const inngestEvents = { manifestCompleted: defineEvent("manifest/completed"), manifestFailed: defineEvent("manifest/failed"), notificationRequested: defineEvent("notification/requested"), + serviceCronExecute: defineEvent("service-cron/execute"), }; diff --git a/web/lib/inngest/events/service-cron.ts b/web/lib/inngest/events/service-cron.ts new file mode 100644 index 00000000..49fc3e4a --- /dev/null +++ b/web/lib/inngest/events/service-cron.ts @@ -0,0 +1,5 @@ +export type ServiceCronEvents = { + "service-cron/execute": { + data: { cronId: string; schedule: string; scheduledFor: string }; + }; +}; diff --git a/web/lib/inngest/functions/crons.ts b/web/lib/inngest/functions/crons.ts index ddcd433b..89b5a9b7 100644 --- a/web/lib/inngest/functions/crons.ts +++ b/web/lib/inngest/functions/crons.ts @@ -1,4 +1,7 @@ +import { and, asc, eq, isNull, lte } from "drizzle-orm"; import { cron } from "inngest"; +import { db } from "@/db"; +import { serviceCrons, services } from "@/db/schema"; import { cleanupExpiredChallenges, renewExpiringCertificates, @@ -19,6 +22,11 @@ import { runAutoscalingController, } from "@/lib/scheduler"; import { inngest } from "../client"; +import { + cronEventId, + latestDueOccurrence, + nextOccurrenceAfter, +} from "@/lib/service-crons"; export const staleServerCheck = inngest.createFunction( { @@ -206,3 +214,58 @@ export const serviceCommandRetention = inngest.createFunction( async ({ step }) => step.run("cleanup-service-commands", cleanupOldServiceCommands), ); + +export const serviceCronDispatcher = inngest.createFunction( + { + id: "cron-service-cron-dispatcher", + triggers: [cron("* * * * *")], + singleton: { mode: "skip" }, + }, + async ({ step }) => + step.run("dispatch-service-crons", async () => { + const now = new Date(); + const rows = await db + .select({ cron: serviceCrons }) + .from(serviceCrons) + .innerJoin( + services, + and( + eq(serviceCrons.serviceId, services.id), + isNull(services.deletedAt), + ), + ) + .where(lte(serviceCrons.nextScheduledFor, now)) + .orderBy(asc(serviceCrons.nextScheduledFor)) + .limit(100) + .then((results) => results.map(({ cron }) => cron)); + for (const row of rows) { + const occurrence = latestDueOccurrence( + row.schedule, + new Date(row.nextScheduledFor.getTime() - 1), + now, + ); + if (!occurrence) continue; + await inngest.send({ + id: cronEventId(row.id, occurrence), + name: "service-cron/execute", + data: { + cronId: row.id, + schedule: row.schedule, + scheduledFor: occurrence.toISOString(), + }, + }); + await db + .update(serviceCrons) + .set({ + lastScheduledFor: occurrence, + nextScheduledFor: nextOccurrenceAfter(row.schedule, now), + }) + .where( + and( + eq(serviceCrons.id, row.id), + eq(serviceCrons.nextScheduledFor, row.nextScheduledFor), + ), + ); + } + }), +); diff --git a/web/lib/inngest/functions/index.ts b/web/lib/inngest/functions/index.ts index 51caadfa..88a7b203 100644 --- a/web/lib/inngest/functions/index.ts +++ b/web/lib/inngest/functions/index.ts @@ -9,6 +9,7 @@ export { controlPlaneUpdateCheck, notificationRetention, serviceCommandRetention, + serviceCronDispatcher, oldBackupsCleanup, registryArtifactRetention, scheduledBackupsCheck, @@ -16,6 +17,7 @@ export { staleItemsCleanup, staleServerCheck, } from "./crons"; +export { serviceCronWorkflow } from "./service-cron-workflow"; export { migrationWorkflow } from "./migration-workflow"; export { notificationDelivery } from "./notification-delivery"; export { onDeploymentFailed } from "./on-deployment-failed"; diff --git a/web/lib/inngest/functions/service-cron-workflow.ts b/web/lib/inngest/functions/service-cron-workflow.ts new file mode 100644 index 00000000..3bec2513 --- /dev/null +++ b/web/lib/inngest/functions/service-cron-workflow.ts @@ -0,0 +1,20 @@ +import { executeServiceCron } from "@/lib/service-crons"; +import { inngest } from "../client"; +import { inngestEvents } from "../events"; + +export const serviceCronWorkflow = inngest.createFunction( + { + id: "service-cron-execute", + retries: 0, + concurrency: [{ limit: 1, key: "event.data.cronId" }], + triggers: [inngestEvents.serviceCronExecute], + }, + async ({ event, step }) => + step.run("execute-service-cron", () => + executeServiceCron( + event.data.cronId, + event.data.schedule, + new Date(event.data.scheduledFor), + ), + ), +); diff --git a/web/lib/public-api-routes.ts b/web/lib/public-api-routes.ts index 66234516..2ea2f759 100644 --- a/web/lib/public-api-routes.ts +++ b/web/lib/public-api-routes.ts @@ -133,6 +133,7 @@ export async function getServiceDetails( startCommand: current.startCommand, resources: current.resources.cpuCores == null ? null : current.resources, + crons: current.crons, }, management: configuration.management, }); diff --git a/web/lib/public-api.ts b/web/lib/public-api.ts index a6159dce..e9993ea1 100644 --- a/web/lib/public-api.ts +++ b/web/lib/public-api.ts @@ -1,5 +1,15 @@ import { createHash, randomUUID } from "node:crypto"; -import { and, desc, eq, inArray, isNull, ne, sql } from "drizzle-orm"; +import { + and, + desc, + eq, + inArray, + isNull, + ne, + notInArray, + sql, +} from "drizzle-orm"; +import { CronExpressionParser } from "cron-parser"; import { z } from "zod"; import { db } from "@/db"; import { @@ -8,6 +18,7 @@ import { githubRepos, projects, servers, + serviceCrons, servicePorts, serviceReplicas, serviceRevisions, @@ -27,6 +38,59 @@ import { const githubPathPart = /^[A-Za-z0-9_.-]+$/; const windowsAbsolutePath = /^[A-Za-z]:[\\/]/; +export function nextOccurrenceAfter(schedule: string, after: Date): Date { + return CronExpressionParser.parse(schedule, { + currentDate: after, + tz: "UTC", + }) + .next() + .toDate(); +} + +export function isSafeCronPath(value: string): boolean { + if (!value.startsWith("/") || value.startsWith("//") || value.length > 2048) + return false; + let decoded: string; + try { + decoded = decodeURIComponent(value); + } catch { + return false; + } + if ( + !decoded.startsWith("/") || + decoded.startsWith("//") || + // eslint-disable-next-line no-control-regex -- Manifest paths must reject control characters. + /[?#\\\u0000-\u001f\u007f]/.test(decoded) + ) + return false; + return !decoded.split("/").some((part) => part === "." || part === ".."); +} + +export const publicCronSchema = z.strictObject({ + path: z + .string() + .trim() + .min(1) + .max(2048) + .refine(isSafeCronPath, "Invalid cron path"), + schedule: z + .string() + .trim() + .min(1) + .max(255) + .refine((value) => value.split(/\s+/).length === 5, { + message: "Schedule must contain exactly five cron fields", + }) + .refine((value) => { + try { + CronExpressionParser.parse(value, { tz: "UTC" }); + return true; + } catch { + return false; + } + }, "Invalid UTC cron expression"), +}); + export function canonicalGitHubRepository(value: string): string { let url: URL; try { @@ -314,7 +378,7 @@ function sanitizeSpec(specification: unknown) { } export async function safeConfiguration(service: NestedService) { - const [repo, ports, volumes, placements, activeDeployment] = + const [repo, ports, crons, volumes, placements, activeDeployment] = await Promise.all([ db .select() @@ -326,6 +390,10 @@ export async function safeConfiguration(service: NestedService) { .select() .from(servicePorts) .where(eq(servicePorts.serviceId, service.id)), + db + .select({ path: serviceCrons.path, schedule: serviceCrons.schedule }) + .from(serviceCrons) + .where(eq(serviceCrons.serviceId, service.id)), db .select({ name: serviceVolumes.name, @@ -374,6 +442,9 @@ export async function safeConfiguration(service: NestedService) { a.name.localeCompare(b.name, "en") || a.containerPath.localeCompare(b.containerPath, "en"), ); + const sortedCrons = crons.toSorted((a, b) => + a.path.localeCompare(b.path, "en"), + ); const replicaCount = getServiceTotalReplicas({ ...service, configuredReplicas: sortedPlacements, @@ -428,6 +499,7 @@ export async function safeConfiguration(service: NestedService) { tlsPassthrough: port.tlsPassthrough, })), volumes: sortedVolumes, + crons: sortedCrons, serverless: { enabled: service.serverlessEnabled, sleepAfterSeconds: service.serverlessSleepAfterSeconds, @@ -628,6 +700,16 @@ export const replaceConfigurationSchema = z.strictObject({ "CPU and memory limits must both be set or both be null", ) .nullable(), + crons: z + .array(publicCronSchema) + .max(100) + .superRefine((crons, context) => { + if (new Set(crons.map((cron) => cron.path)).size !== crons.length) + context.addIssue({ + code: "custom", + message: "Cron paths must be unique", + }); + }), }); type PublicApiDomainError = Error & { code: string; status: number }; @@ -664,6 +746,9 @@ function healthCheckFromService(service: NestedService) { } type ReplacementInput = z.infer; +type CanonicalReplacementInput = Omit & { + crons?: ReplacementInput["crons"]; +}; type ConfigurationChange = { field: string; from: unknown; to: unknown }; function canonicalPlanSource(source: PublicSource) { @@ -680,6 +765,7 @@ function canonicalReplacementState( source: ReturnType, ports: Array<{ port: number; isPublic: boolean; domain: string | null }>, placements: Array<{ serverId: string; count: number }>, + crons: Array<{ path: string; schedule: string }>, ) { const resources = service.resourceCpuLimit == null && service.resourceMemoryLimitMb == null @@ -726,11 +812,14 @@ function canonicalReplacementState( healthCheck: healthCheckFromService(service), startCommand: service.startCommand?.trim() || null, resources, + crons: crons + .map(({ path, schedule }) => ({ path, schedule })) + .toSorted((a, b) => a.path.localeCompare(b.path, "en")), serverless: { enabled: service.serverlessEnabled }, }; } -export function canonicalDesired(input: ReplacementInput) { +export function canonicalDesired(input: CanonicalReplacementInput) { return { ...input, source: canonicalPlanSource(input.source), @@ -762,6 +851,9 @@ export function canonicalDesired(input: ReplacementInput) { }, } : input.placement, + crons: (input.crons ?? []).toSorted((a, b) => + a.path.localeCompare(b.path, "en"), + ), }; } @@ -807,12 +899,19 @@ function configurationChanges( } export function planCanonicalConfiguration( - current: ReturnType, - desiredInput: ReplacementInput, + current: Omit, "crons"> & { + crons?: Array<{ path: string; schedule: string }>; + }, + desiredInput: CanonicalReplacementInput, ) { + const { serverless, ...currentWithoutServerless } = current; const canonicalCurrent = { - ...current, + ...currentWithoutServerless, source: canonicalPlanSource(current.source), + crons: (current.crons ?? []) + .map(({ path, schedule }) => ({ path, schedule })) + .toSorted((a, b) => a.path.localeCompare(b.path, "en")), + serverless, }; const desiredConfiguration = canonicalDesired(desiredInput); const desired = { @@ -880,7 +979,7 @@ async function replaceConfigurationInternal( .limit(1) .then((rows) => rows[0]); if (!persisted) domainError("Service not found", "NOT_FOUND", 404); - const [ports, volumes, placements, repo] = await Promise.all([ + const [ports, volumes, placements, repo, crons] = await Promise.all([ tx .select() .from(servicePorts) @@ -899,6 +998,10 @@ async function replaceConfigurationInternal( .where(eq(githubRepos.serviceId, service.id)) .limit(1) .then((rows) => rows[0]), + tx + .select() + .from(serviceCrons) + .where(eq(serviceCrons.serviceId, service.id)), ]); const source = resolvePersistedSourceFromRows(persisted, repo); const currentState = canonicalReplacementState( @@ -906,6 +1009,7 @@ async function replaceConfigurationInternal( source, ports, placements, + crons, ); if ( input.source.type === "image" && @@ -1300,6 +1404,42 @@ async function replaceConfigurationInternal( } } } + const cronsByPath = new Map(crons.map((cron) => [cron.path, cron])); + const appliedAt = new Date(); + for (const cron of input.crons) { + const existing = cronsByPath.get(cron.path); + if (!existing) { + await tx.insert(serviceCrons).values({ + id: randomUUID(), + serviceId: service.id, + ...cron, + nextScheduledFor: nextOccurrenceAfter(cron.schedule, appliedAt), + }); + } else if (existing.schedule !== cron.schedule) { + await tx + .update(serviceCrons) + .set({ + schedule: cron.schedule, + nextScheduledFor: nextOccurrenceAfter(cron.schedule, appliedAt), + }) + .where(eq(serviceCrons.id, existing.id)); + } + } + if (input.crons.length === 0) { + await tx + .delete(serviceCrons) + .where(eq(serviceCrons.serviceId, service.id)); + } else { + await tx.delete(serviceCrons).where( + and( + eq(serviceCrons.serviceId, service.id), + notInArray( + serviceCrons.path, + input.crons.map((cron) => cron.path), + ), + ), + ); + } return { targetServiceName: persisted.name, ...plan }; }); diff --git a/web/lib/service-crons.ts b/web/lib/service-crons.ts new file mode 100644 index 00000000..78d0d8db --- /dev/null +++ b/web/lib/service-crons.ts @@ -0,0 +1,362 @@ +import { lookup as dnsLookup } from "node:dns/promises"; +import * as http from "node:http"; +import * as https from "node:https"; +import { isIP } from "node:net"; +import { and, eq, inArray, isNull, lt, or } from "drizzle-orm"; +import { CronExpressionParser } from "cron-parser"; +import { Address4, Address6 } from "ip-address"; +import { db } from "@/db"; +import { secrets, serviceCrons, services } from "@/db/schema"; +import { decryptSecret } from "@/lib/crypto"; +import { isSafeCronPath, nextOccurrenceAfter } from "@/lib/public-api"; +import { ingestCronLog, type CronLog } from "@/lib/victoria-logs"; + +const MAX_ERROR = 500; +const EXECUTION_BUDGET_MS = 10_000; +const blockedV4 = [ + "0.0.0.0/8", + "10.0.0.0/8", + "100.64.0.0/10", + "127.0.0.0/8", + "169.254.0.0/16", + "172.16.0.0/12", + "192.0.0.0/24", + "192.0.2.0/24", + "192.88.99.0/24", + "192.168.0.0/16", + "198.18.0.0/15", + "198.51.100.0/24", + "203.0.113.0/24", + "224.0.0.0/4", + "240.0.0.0/4", +].map((value) => new Address4(value)); +const blockedV6 = ["2001::/23", "2001:db8::/32", "3fff::/20"].map( + (value) => new Address6(value), +); + +export type ResolvedAddress = { address: string; family: 4 | 6 }; +export type CronRequestResult = { + status: "succeeded" | "failed"; + statusCode: number | null; + error: string | null; +}; + +export { nextOccurrenceAfter }; + +export function latestDueOccurrence( + schedule: string, + cursor: Date, + now: Date, +): Date | null { + try { + const occurrence = CronExpressionParser.parse(schedule, { + currentDate: new Date(now.getTime() + 1), + tz: "UTC", + }) + .prev() + .toDate(); + return occurrence > cursor && occurrence <= now ? occurrence : null; + } catch { + return null; + } +} + +export function cronEventId(cronId: string, scheduledFor: Date): string { + return `service-cron:${cronId}:${scheduledFor.toISOString()}`; +} + +export function sanitizeCronError(error: unknown): string { + const message = + error instanceof Error ? error.message : "Cron request failed"; + // eslint-disable-next-line no-control-regex -- Strip unsafe control characters from persisted errors. + return message.replace(/[\u0000-\u001f\u007f]/g, " ").slice(0, MAX_ERROR); +} + +export function isGlobalAddress(value: string): boolean { + try { + if (isIP(value) === 4) { + const address = new Address4(value); + return !blockedV4.some((range) => address.isInSubnet(range)); + } + if (isIP(value) === 6) { + const address = new Address6(value); + if (address.is4()) return isGlobalAddress(address.to4().address); + return ( + address.isInSubnet(new Address6("2000::/3")) && + !blockedV6.some((range) => address.isInSubnet(range)) + ); + } + } catch {} + return false; +} + +export function parseCronUrl(base: string, path: string): URL { + if (!isSafeCronPath(path)) throw new Error("Invalid cron path"); + let url: URL; + try { + url = new URL(base); + } catch { + throw new Error("Invalid CRON_BASE_URL"); + } + if ( + (url.protocol !== "http:" && url.protocol !== "https:") || + url.username || + url.password || + url.search || + url.hash + ) + throw new Error("Invalid CRON_BASE_URL"); + return new URL(path, `${url.origin}/`); +} + +export async function resolvePublicAddresses( + hostname: string, + lookup: ( + hostname: string, + options: { all: true; verbatim: true }, + ) => Promise = async (host, options) => + (await dnsLookup(host, options)).map((answer) => ({ + address: answer.address, + family: answer.family as 4 | 6, + })), +): Promise { + const literal = isIP(hostname); + let answers: ResolvedAddress[]; + try { + answers = literal + ? [{ address: hostname, family: literal as 4 | 6 }] + : await lookup(hostname, { all: true, verbatim: true }); + } catch { + throw new Error("DNS lookup failed"); + } + if ( + !answers.length || + answers.some(({ address }) => !isGlobalAddress(address)) + ) + throw new Error("Cron destination is not public"); + return answers; +} + +type RequestImpl = typeof http.request; +export function validateCronTransport(url: URL, secret?: string): void { + if (secret && url.protocol === "http:") + throw new Error("CRON_SECRET requires HTTPS"); +} + +export async function performCronGet( + url: URL, + addresses: ResolvedAddress[], + secret: string | undefined, + timeoutMs: number, + requestImpl: RequestImpl = url.protocol === "https:" + ? https.request + : http.request, +): Promise { + return new Promise((resolve) => { + let settled = false; + const finish = (result: CronRequestResult) => { + if (settled) return; + settled = true; + resolve(result); + }; + if (timeoutMs <= 0) + return finish({ + status: "failed", + statusCode: null, + error: "Cron request timed out", + }); + const requestOptions: http.RequestOptions & { autoSelectFamily: boolean } = + { + method: "GET", + agent: false, + autoSelectFamily: false, + headers: secret ? { Authorization: `Bearer ${secret}` } : undefined, + lookup: (_hostname, options, callback) => { + const selected = addresses[0]; + if (options?.all) callback(null, addresses); + else callback(null, selected.address, selected.family); + }, + ...(url.protocol === "https:" ? { servername: url.hostname } : {}), + }; + const req = requestImpl(url, requestOptions, (response) => { + const code = response.statusCode ?? null; + response.destroy(); + finish( + code != null && code >= 200 && code < 300 + ? { status: "succeeded", statusCode: code, error: null } + : { + status: "failed", + statusCode: code, + error: `HTTP status ${code ?? "unknown"}`, + }, + ); + }); + const timer = setTimeout(() => { + finish({ + status: "failed", + statusCode: null, + error: "Cron request timed out", + }); + req.destroy(); + }, timeoutMs); + timer.unref?.(); + req.once("error", () => + finish({ + status: "failed", + statusCode: null, + error: "Cron request failed", + }), + ); + req.once("close", () => { + clearTimeout(timer); + if (!settled) + finish({ + status: "failed", + statusCode: null, + error: "Cron request closed", + }); + }); + req.end(); + }); +} + +async function withinDeadline( + promise: Promise, + deadline: number, +): Promise { + const remaining = deadline - Date.now(); + if (remaining <= 0) throw new Error("Cron request timed out"); + let timer: ReturnType; + try { + return await Promise.race([ + promise, + new Promise((_, reject) => { + timer = setTimeout( + () => reject(new Error("Cron request timed out")), + remaining, + ); + }), + ]); + } finally { + clearTimeout(timer!); + } +} + +export async function executeServiceCron( + cronId: string, + schedule: string, + scheduledFor: Date, +) { + const deadline = Date.now() + EXECUTION_BUDGET_MS; + const row = await db + .select({ cron: serviceCrons, serviceId: services.id }) + .from(serviceCrons) + .innerJoin( + services, + and(eq(serviceCrons.serviceId, services.id), isNull(services.deletedAt)), + ) + .where(eq(serviceCrons.id, cronId)) + .limit(1) + .then((rows) => rows[0]); + if (!row || row.cron.schedule !== schedule) return { stale: true as const }; + const startedAt = new Date(); + const claimed = await db + .update(serviceCrons) + .set({ lastAttemptedFor: scheduledFor, lastStartedAt: startedAt }) + .where( + and( + eq(serviceCrons.id, cronId), + eq(serviceCrons.schedule, schedule), + or( + isNull(serviceCrons.lastAttemptedFor), + lt(serviceCrons.lastAttemptedFor, scheduledFor), + ), + ), + ) + .returning({ id: serviceCrons.id }); + if (!claimed.length) return { stale: true as const }; + let status: "succeeded" | "failed" | "skipped" = "skipped"; + let statusCode: number | null = null; + let error: string | null = null; + let base = ""; + let secret: string | undefined; + try { + const values = await db + .select() + .from(secrets) + .where( + and( + eq(secrets.serviceId, row.serviceId), + inArray(secrets.key, ["CRON_BASE_URL", "CRON_SECRET"]), + ), + ); + const encrypted = new Map( + values.map((value) => [value.key, value.encryptedValue]), + ); + base = encrypted.get("CRON_BASE_URL") + ? await decryptSecret(encrypted.get("CRON_BASE_URL")!) + : ""; + secret = encrypted.get("CRON_SECRET") + ? await decryptSecret(encrypted.get("CRON_SECRET")!) + : undefined; + } catch { + status = "failed"; + error = "Cron configuration could not be loaded"; + } + if (error === null) { + try { + if (!base.trim()) throw new Error("CRON_BASE_URL is not configured"); + const url = parseCronUrl(base.trim(), row.cron.path); + validateCronTransport(url, secret); + try { + const addresses = await withinDeadline( + resolvePublicAddresses(url.hostname), + deadline, + ); + ({ status, statusCode, error } = await performCronGet( + url, + addresses, + secret, + deadline - Date.now(), + )); + } catch (cause) { + const message = sanitizeCronError(cause); + status = + message === "Cron destination is not public" ? "skipped" : "failed"; + error = message; + } + } catch (cause) { + error = sanitizeCronError(cause); + status = "skipped"; + } + } + const finishedAt = new Date(); + const durationMs = Math.max(0, finishedAt.getTime() - startedAt.getTime()); + await db + .update(serviceCrons) + .set({ + lastFinishedAt: finishedAt, + lastStatus: status, + lastStatusCode: statusCode, + lastDurationMs: durationMs, + lastError: error, + }) + .where(eq(serviceCrons.id, cronId)); + const log: CronLog = { + _msg: `Cron ${status}`, + _time: finishedAt.toISOString(), + service_id: row.serviceId, + cron_id: cronId, + path: row.cron.path, + scheduled_for: scheduledFor.toISOString(), + started_at: startedAt.toISOString(), + finished_at: finishedAt.toISOString(), + result: status, + status: statusCode, + duration_ms: durationMs, + error, + log_type: "cron", + }; + await ingestCronLog(log); + return { stale: false as const, status, statusCode, error }; +} diff --git a/web/lib/victoria-logs.ts b/web/lib/victoria-logs.ts index 6fd293f3..482e5f6a 100644 --- a/web/lib/victoria-logs.ts +++ b/web/lib/victoria-logs.ts @@ -20,7 +20,7 @@ function getQueryEndpoint(): EndpointConfig | undefined { return parseEndpoint(endpoint); } -export type LogType = "container" | "http"; +export type LogType = "container" | "http" | "cron"; type LogSearchField = "_msg" | "path" | "method" | "status" | "client_ip"; export type StoredLog = { @@ -114,8 +114,10 @@ function buildServiceLogFilter(options: QueryLogsByServiceOptions): string { let query = formatLogSqlExactFilter("service_id", serviceId); if (logType === "http") { query += ` log_type:http`; + } else if (logType === "cron") { + query += ` log_type:cron`; } else if (logType === "container") { - query += ` -log_type:http -log_type:build -log_type:rollout`; + query += ` -log_type:http -log_type:build -log_type:rollout -log_type:cron`; } else { query += ` -log_type:build -log_type:rollout`; } @@ -354,6 +356,42 @@ export type RolloutLog = { log_type: "rollout"; }; +export type CronLog = { + _msg: string; + _time: string; + service_id: string; + cron_id: string; + path: string; + scheduled_for: string; + started_at: string; + finished_at: string; + result: "succeeded" | "failed" | "skipped"; + status: number | null; + duration_ms: number; + error: string | null; + log_type: "cron"; +}; + +export async function ingestCronLog(entry: CronLog): Promise { + try { + const endpoint = getQueryEndpoint(); + if (!endpoint) return; + const options = buildFetchOptions(endpoint); + await fetch(`${endpoint.url}/insert/jsonline`, { + ...options, + method: "POST", + body: `${JSON.stringify(entry)}\n`, + headers: { + ...((options.headers as Record) || {}), + "Content-Type": "application/json", + }, + signal: AbortSignal.timeout(5_000), + }); + } catch { + // Execution history is best-effort and must not affect the request result. + } +} + export async function ingestRolloutLog( rolloutId: string, serviceId: string, diff --git a/web/tests/inngest-route.test.ts b/web/tests/inngest-route.test.ts index 316bd55d..e532acea 100644 --- a/web/tests/inngest-route.test.ts +++ b/web/tests/inngest-route.test.ts @@ -28,6 +28,8 @@ const mocks = vi.hoisted(() => { scheduledBackupsCheck: { id: "scheduled-backups-check" }, scheduledDeploymentsCheck: { id: "scheduled-deployments-check" }, serviceCommandRetention: { id: "service-command-retention" }, + serviceCronDispatcher: { id: "service-cron-dispatcher" }, + serviceCronWorkflow: { id: "service-cron-workflow" }, serviceDeletionWorkflow: { id: "service-deletion-workflow" }, serviceRestoreWorkflow: { id: "service-restore-workflow" }, staleItemsCleanup: { id: "stale-items-cleanup" }, diff --git a/web/tests/log-routes.test.ts b/web/tests/log-routes.test.ts index 144c1031..e53f9146 100644 --- a/web/tests/log-routes.test.ts +++ b/web/tests/log-routes.test.ts @@ -69,6 +69,18 @@ describe("log routes", () => { }); }); + it("accepts cron as a service log type", async () => { + const { GET, queryLogsByService } = await loadServiceLogsRoute(); + const response = await GET( + new Request("http://localhost/api/services/service-1/logs?type=cron"), + { params: Promise.resolve({ id: "service-1" }) }, + ); + expect(response.status).toBe(200); + expect(queryLogsByService).toHaveBeenCalledWith( + expect.objectContaining({ logType: "cron" }), + ); + }); + it("passes a validated after cursor to the deployment query", async () => { const { GET, queryLogsByDeployment } = await loadDeploymentLogsRoute(); const cursor = "2026-07-10T01:02:03Z"; diff --git a/web/tests/public-api-configuration.test.ts b/web/tests/public-api-configuration.test.ts index 3d01ccef..ad890425 100644 --- a/web/tests/public-api-configuration.test.ts +++ b/web/tests/public-api-configuration.test.ts @@ -34,6 +34,7 @@ describe("public API configuration state", () => { [], [], [], + [], [{ serverId: "server-1", serverName: "Sydney", count: 1 }], [{ id: "deployment-1", revisionId: "revision-1" }], [ diff --git a/web/tests/public-api-plan.test.ts b/web/tests/public-api-plan.test.ts index c9813780..a1939ee5 100644 --- a/web/tests/public-api-plan.test.ts +++ b/web/tests/public-api-plan.test.ts @@ -135,6 +135,80 @@ describe("configuration plan protocol", () => { ).toEqual(["a", "z"]); }); + it("sorts crons by path and includes schedule changes in the plan", () => { + const desired = canonicalDesired({ + name: "web", + source: { type: "image", image: "nginx" }, + hostname: "web", + ports: [], + placement: { mode: "automatic", replicas: 1 }, + healthCheck: null, + startCommand: null, + resources: null, + crons: [ + { path: "/z", schedule: "0 5 * * *" }, + { path: "/a", schedule: "0 6 * * *" }, + ], + }); + expect(desired.crons.map((cron) => cron.path)).toEqual(["/a", "/z"]); + + const result = planCanonicalConfiguration( + { + ...desired, + serverless: { enabled: false }, + crons: [{ path: "/a", schedule: "0 7 * * *" }], + }, + { ...desired, source: { type: "image", image: "nginx" } }, + ); + expect(result.changes).toContainEqual({ + field: "crons", + from: [{ path: "/a", schedule: "0 7 * * *" }], + to: desired.crons, + }); + }); + + it("excludes cron runtime summaries from configuration identity", () => { + const definition = { path: "/job", schedule: "0 5 * * *" }; + const currentCron = { + ...definition, + id: "cron-1", + lastStatus: "succeeded", + lastStatusCode: 204, + }; + const failedCurrentCron = { + ...currentCron, + lastStatus: "failed", + lastStatusCode: 500, + }; + const base = { + name: "web", + source: { type: "image" as const, image: "nginx" }, + hostname: "web", + ports: [], + placement: { mode: "automatic" as const, replicas: 1 }, + healthCheck: null, + startCommand: null, + resources: null, + }; + + const first = planCanonicalConfiguration( + { ...base, crons: [currentCron], serverless: { enabled: false } }, + { ...base, crons: [definition] }, + ); + const second = planCanonicalConfiguration( + { + ...base, + crons: [failedCurrentCron], + serverless: { enabled: false }, + }, + { ...base, crons: [definition] }, + ); + + expect(first.action).toBe("noop"); + expect(first.changes).toEqual([]); + expect(second.currentVersion).toBe(first.currentVersion); + }); + it("reports every managed field change, including removals and null clears", () => { const current = { name: "old-web", diff --git a/web/tests/public-api-source.test.ts b/web/tests/public-api-source.test.ts index 202f880f..d758ed99 100644 --- a/web/tests/public-api-source.test.ts +++ b/web/tests/public-api-source.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from "vitest"; import { canonicalGitHubRepository, + isSafeCronPath, isSafeRepositoryRoot, publicSourceSchema, replaceConfigurationSchema, @@ -15,6 +16,7 @@ const completeConfiguration = (overrides: Record = {}) => ({ healthCheck: null, startCommand: null, resources: null, + crons: [], ...overrides, }); @@ -194,3 +196,52 @@ describe("public API placement schema", () => { ).toBe(false); }); }); + +describe("public API cron schema", () => { + it.each([ + "//host/job", + "/job?x=1", + "/job#part", + "/job\\next", + "/a/../b", + "/a/%2e%2e/b", + "/%2fhost", + "/bad%zz", + "/line%0Abreak", + ])("rejects unsafe cron path %j", (path) => { + expect(isSafeCronPath(path)).toBe(false); + }); + + it("accepts valid unique five-field UTC cron definitions", () => { + expect( + replaceConfigurationSchema.safeParse( + completeConfiguration({ + crons: [{ path: "/jobs/nightly", schedule: "0 5 * * *" }], + }), + ).success, + ).toBe(true); + }); + + it.each([ + [{ path: "/jobs/nightly", schedule: "0 5 * *" }, "non-five-field schedule"], + [ + { path: "/jobs/nightly", schedule: "99 5 * * *" }, + "invalid cron expression", + ], + ])("rejects %s", (cron) => { + expect( + replaceConfigurationSchema.safeParse( + completeConfiguration({ crons: [cron] }), + ).success, + ).toBe(false); + }); + + it("rejects duplicate paths", () => { + const cron = { path: "/jobs/nightly", schedule: "0 5 * * *" }; + expect( + replaceConfigurationSchema.safeParse( + completeConfiguration({ crons: [cron, cron] }), + ).success, + ).toBe(false); + }); +}); diff --git a/web/tests/service-crons.test.ts b/web/tests/service-crons.test.ts new file mode 100644 index 00000000..351f054e --- /dev/null +++ b/web/tests/service-crons.test.ts @@ -0,0 +1,213 @@ +import { EventEmitter } from "node:events"; +import type { ClientRequest, IncomingMessage, RequestOptions } from "node:http"; +import { describe, expect, it, vi } from "vitest"; +import { + cronEventId, + isGlobalAddress, + latestDueOccurrence, + nextOccurrenceAfter, + parseCronUrl, + performCronGet, + resolvePublicAddresses, + validateCronTransport, +} from "@/lib/service-crons"; + +describe("service cron scheduling and SSRF validation", () => { + it("returns the first UTC occurrence strictly after the supplied instant", () => { + expect( + nextOccurrenceAfter( + "0 5 * * *", + new Date("2026-08-06T05:00:00.000Z"), + ).toISOString(), + ).toBe("2026-08-07T05:00:00.000Z"); + expect( + nextOccurrenceAfter( + "30 23 * * *", + new Date("2026-08-06T22:00:00-04:00"), + ).toISOString(), + ).toBe("2026-08-07T23:30:00.000Z"); + }); + + it("coalesces missed intervals to the latest UTC occurrence", () => { + const due = latestDueOccurrence( + "* * * * *", + new Date("2000-01-01Z"), + new Date("2026-08-06T10:05:30Z"), + ); + expect(due?.toISOString()).toBe("2026-08-06T10:05:00.000Z"); + expect(latestDueOccurrence("* * * * *", due!, due!)).toBeNull(); + expect(cronEventId("cron-1", due!)).toBe(cronEventId("cron-1", due!)); + }); + + it.each([ + ["ftp://example.com", "/job"], + ["https://user:pass@example.com", "/job"], + ["https://example.com?x=1", "/job"], + ["https://example.com#x", "/job"], + ["https://example.com", "//host/job"], + ["https://example.com", "/a/%2e%2e/job"], + ])("rejects unsafe URL %s %s", (base, path) => + expect(() => parseCronUrl(base, path)).toThrow(), + ); + + it("joins a safe path while preserving the origin hostname", () => { + expect(parseCronUrl("https://example.com", "/jobs/nightly").href).toBe( + "https://example.com/jobs/nightly", + ); + }); + + it.each([ + "10.0.0.1", + "127.0.0.1", + "169.254.1.1", + "192.0.2.1", + "192.88.99.1", + "::1", + "fc00::1", + "fe80::1", + "2001::1", + "2001:db8::1", + "3fff::1", + "::ffff:10.0.0.1", + ])("rejects non-global address %s", (address) => + expect(isGlobalAddress(address)).toBe(false), + ); + + it("rejects mixed DNS results", async () => { + const lookup = vi.fn(async () => [ + { address: "8.8.8.8", family: 4 as const }, + { address: "10.0.0.1", family: 4 as const }, + ]); + await expect(resolvePublicAddresses("example.com", lookup)).rejects.toThrow( + "not public", + ); + }); + + it("rejects a secret over cleartext HTTP", () => { + expect(() => + validateCronTransport(new URL("http://example.com/job"), "not-logged"), + ).toThrow("requires HTTPS"); + expect(() => + validateCronTransport(new URL("http://example.com/job")), + ).not.toThrow(); + }); + + it.each([false, true])( + "pins lookup for all=%s without pooling", + async (all) => { + let options!: RequestOptions & { autoSelectFamily?: boolean }; + const request = fakeRequest(204, (received) => { + options = received; + }); + const result = performCronGet( + new URL("https://example.com/job"), + [ + { address: "8.8.8.8", family: 4 }, + { address: "2001:4860:4860::8888", family: 6 }, + ], + undefined, + 1_000, + request, + ); + const callback = vi.fn(); + options.lookup!("example.com", { all }, callback); + if (all) + expect(callback).toHaveBeenCalledWith(null, [ + { address: "8.8.8.8", family: 4 }, + { address: "2001:4860:4860::8888", family: 6 }, + ]); + else expect(callback).toHaveBeenCalledWith(null, "8.8.8.8", 4); + expect(options).toMatchObject({ + agent: false, + autoSelectFamily: false, + servername: "example.com", + }); + expect(await result).toEqual({ + status: "succeeded", + statusCode: 204, + error: null, + }); + expect(request).toHaveBeenCalledTimes(1); + }, + ); + + it.each([302, 404])( + "classifies HTTP %s at headers without retaining a body", + async (code) => { + let destroyed = false; + const request = fakeRequest(code, undefined, () => { + destroyed = true; + }); + await expect( + performCronGet( + new URL("https://example.com/job"), + [{ address: "8.8.8.8", family: 4 }], + undefined, + 1_000, + request, + ), + ).resolves.toMatchObject({ status: "failed", statusCode: code }); + expect(destroyed).toBe(true); + expect(request).toHaveBeenCalledTimes(1); + }, + ); + + it("settles once when the absolute request timeout expires", async () => { + vi.useFakeTimers(); + try { + const request = vi.fn(() => { + const req = new EventEmitter() as ClientRequest; + req.end = vi.fn(() => req) as unknown as ClientRequest["end"]; + req.destroy = vi.fn(() => req) as ClientRequest["destroy"]; + return req; + }) as unknown as typeof import("node:http").request; + const result = performCronGet( + new URL("https://example.com/job"), + [{ address: "8.8.8.8", family: 4 }], + undefined, + 25, + request, + ); + await vi.advanceTimersByTimeAsync(25); + await expect(result).resolves.toEqual({ + status: "failed", + statusCode: null, + error: "Cron request timed out", + }); + expect(request).toHaveBeenCalledTimes(1); + } finally { + vi.useRealTimers(); + } + }); +}); + +function fakeRequest( + statusCode: number, + onOptions?: ( + options: RequestOptions & { autoSelectFamily?: boolean }, + ) => void, + onResponseDestroy?: () => void, +) { + return vi.fn( + ( + _url: URL, + options: RequestOptions, + callback: (response: IncomingMessage) => void, + ) => { + onOptions?.(options); + const req = new EventEmitter() as ClientRequest; + req.end = vi.fn(() => { + const response = new EventEmitter() as IncomingMessage; + response.statusCode = statusCode; + response.destroy = vi.fn(() => { + onResponseDestroy?.(); + return response; + }) as IncomingMessage["destroy"]; + callback(response); + return req; + }) as unknown as ClientRequest["end"]; + req.destroy = vi.fn(() => req) as ClientRequest["destroy"]; + return req; + }, + ) as unknown as typeof import("node:http").request; +} diff --git a/web/tests/victoria-logs.test.ts b/web/tests/victoria-logs.test.ts index 4e2b62f3..070b8a03 100644 --- a/web/tests/victoria-logs.test.ts +++ b/web/tests/victoria-logs.test.ts @@ -356,6 +356,66 @@ describe("VictoriaLogs queries", () => { expect(query).toContain('_msg:~"(?i)connection lost"'); } }); + + it("isolates cron, HTTP, and container service log filters", async () => { + const { queryLogsByService } = await loadVictoriaLogs(); + const queries: string[] = []; + vi.stubGlobal( + "fetch", + vi.fn(async (input: string | URL | Request) => { + queries.push(new URL(String(input)).searchParams.get("query") || ""); + return jsonLinesResponse([]); + }), + ); + await queryLogsByService({ + serviceId: "service-1", + limit: 1, + logType: "cron", + }); + await queryLogsByService({ + serviceId: "service-1", + limit: 1, + logType: "http", + }); + await queryLogsByService({ + serviceId: "service-1", + limit: 1, + logType: "container", + }); + expect(queries[0]).toContain("log_type:cron"); + expect(queries[1]).toContain("log_type:http"); + expect(queries[2]).toContain("-log_type:cron"); + }); + + it("ingests only supplied cron metadata with a five-second deadline", async () => { + const { ingestCronLog } = await loadVictoriaLogs(); + const fetchMock = vi.fn( + async (_input: string | URL | Request, _init?: RequestInit) => + new Response(null, { status: 204 }), + ); + vi.stubGlobal("fetch", fetchMock); + await ingestCronLog({ + _msg: "Cron succeeded", + _time: "2026-08-06T10:00:01Z", + service_id: "service-1", + cron_id: "cron-1", + path: "/job", + scheduled_for: "2026-08-06T10:00:00Z", + started_at: "2026-08-06T10:00:00Z", + finished_at: "2026-08-06T10:00:01Z", + result: "succeeded", + status: 204, + duration_ms: 1000, + error: null, + log_type: "cron", + }); + const [url, init] = fetchMock.mock.calls[0]!; + expect(String(url)).toBe("http://victoria.test/insert/jsonline"); + expect(init?.signal).toBeInstanceOf(AbortSignal); + expect(init?.body).toContain('"status":204'); + expect(init?.body).not.toContain("Authorization"); + expect(init?.body).not.toContain("CRON_BASE_URL"); + }); }); async function loadVictoriaLogs() {