Skip to content

Commit b1f927d

Browse files
DaniAkashDaniAkash
authored
feat: move scheduled jobs and run history to the server (#2538)
* feat(server): add local storage for scheduled job runs Job definitions had a table; their run history did not, so it was the one part of the domain with nowhere to live on this side. Runs cascade on job delete, unlike the job to provider reference which is set null. A job whose provider was removed is a job needing attention, whereas a run whose job was removed means nothing, and deleting a job already removed its runs before this table existed. The tool call log is a json column. Its input field is optional here where the extension has it required: an unknown already admits undefined, so the two describe the same values, and matching the validator avoids asserting the difference away at the route boundary. * feat(server): carry the per-job run cap across with the runs The extension kept fifteen runs per job, trimming as it created each one. Now that it no longer owns the history that policy has to live here, or the table grows without bound. It applies on every write rather than only on creation, which is bounded and idempotent, so it holds however the run was written. The import path does not prune, staying purely additive; the next real run trims. * feat(app): read and write scheduled jobs and runs through the server The hooks keep their shape, so the tasks page, the results view, the card and the new tab panel are unchanged apart from where they import from. Both gain an unavailable state, since an empty list and an unreachable server are now the same shape without one. The alarm runner distinguishes them everywhere it reads. Treating a failed load as an empty list would read as nothing being scheduled: alarms would not be rebuilt on startup and schedules would quietly stop firing, with no failed run to show for it. It skips the pass instead and retries on the next startup. Extension storage no longer carries the data, but it still carries the change signal. Runs are written by the background while the side panel and new tab display them, and storage watch is what kept those in step. A revision item is bumped after a write so every mounted view refetches. Run history is imported once, under its own marker. It cannot share the provider and job marker because that import must never run twice: extension storage is frozen now, so a second pass would insert back whatever the user has since deleted. Also removes the scheduled job deletion queue, whose only reader went when sync did, and the mount-time storage read that chose the opening tab, which is now derived so it settles when the history arrives. * fix(app): record a finished run against the current job Recording that a run finished wrote back the job as it was read before the run started. A run can take minutes and the job stays editable throughout, so a rename, a schedule change, a disable or a different provider chosen while it was going would be silently reverted. The old code merged into a freshly read list; passing the job object instead was an attempt to save a read and is what lost the update. It takes an id again, so a stale snapshot cannot be handed to it, and it skips the write when the job was deleted mid-run rather than resurrecting it.
1 parent d39ed27 commit b1f927d

26 files changed

Lines changed: 2246 additions & 252 deletions

‎packages/browseros-agent/apps/app/entrypoints/background/scheduledJobRuns.ts‎

Lines changed: 57 additions & 59 deletions
Original file line numberDiff line numberDiff line change
@@ -1,45 +1,51 @@
11
import { onScheduleMessage } from '@/lib/messaging/schedules/scheduleMessages'
22
import { createAlarmFromJob } from '@/lib/schedules/createAlarmFromJob'
33
import { getChatServerResponse } from '@/lib/schedules/getChatServerResponse'
4-
import {
5-
scheduledJobRunStorage,
6-
scheduledJobStorage,
7-
} from '@/lib/schedules/scheduleStorage'
84
import type { ScheduledJobRun } from '@/lib/schedules/scheduleTypes'
5+
import {
6+
listScheduledJobRunsOrNull,
7+
listScheduledJobsOrNull,
8+
putScheduledJob,
9+
putScheduledJobRun,
10+
} from '@/modules/schedules/schedules.api'
11+
import { applyLastRunAt } from '@/modules/schedules/schedules.helpers'
912

10-
const MAX_RUNS_PER_JOB = 15
1113
const STALE_TIMEOUT_MS = 10 * 60 * 1000 // 10 minutes
1214
const TWENTY_FOUR_HOURS_MS = 24 * 60 * 60 * 1000
1315

1416
const runAbortControllers = new Map<string, AbortController>()
1517

1618
export const scheduledJobRuns = async () => {
19+
// Every read below distinguishes an unreachable server from an empty list.
20+
// Treating the two alike would look like "nothing is scheduled": alarms would
21+
// not be rebuilt on startup and schedules would quietly stop firing, with no
22+
// failed run to show for it. Skipping the pass instead leaves the next
23+
// startup to retry.
1724
const cleanupStaleJobRuns = async () => {
18-
const current = (await scheduledJobRunStorage.getValue()) ?? []
25+
const current = await listScheduledJobRunsOrNull()
26+
if (current === null) return
1927
const now = Date.now()
2028

21-
const updated = current.map((run) => {
22-
if (run.status !== 'running') return run
23-
24-
const startedAt = new Date(run.startedAt).getTime()
25-
if (now - startedAt > STALE_TIMEOUT_MS) {
26-
return {
27-
...run,
28-
status: 'failed' as const,
29-
completedAt: new Date().toISOString(),
30-
result: 'Job timed out!',
31-
}
32-
}
33-
return run
34-
})
29+
const stale = current.filter(
30+
(run) =>
31+
run.status === 'running' &&
32+
now - new Date(run.startedAt).getTime() > STALE_TIMEOUT_MS,
33+
)
3534

36-
await scheduledJobRunStorage.setValue(updated)
35+
for (const run of stale) {
36+
await putScheduledJobRun({
37+
...run,
38+
status: 'failed',
39+
completedAt: new Date().toISOString(),
40+
result: 'Job timed out!',
41+
})
42+
}
3743
}
3844

3945
const syncAlarmState = async () => {
40-
const jobs = (await scheduledJobStorage.getValue()).filter(
41-
(each) => each.enabled,
42-
)
46+
const loaded = await listScheduledJobsOrNull()
47+
if (loaded === null) return
48+
const jobs = loaded.filter((each) => each.enabled)
4349

4450
for (let i = 0; i < jobs.length; i++) {
4551
const job = jobs[i]
@@ -56,55 +62,46 @@ export const scheduledJobRuns = async () => {
5662
jobId: string,
5763
status: ScheduledJobRun['status'],
5864
): Promise<ScheduledJobRun> => {
65+
// Trimming to the per-job cap happens on the server now, so creating a run
66+
// no longer has to rewrite the job's whole history to stay bounded.
5967
const jobRun: ScheduledJobRun = {
6068
id: crypto.randomUUID(),
6169
jobId,
6270
startedAt: new Date().toISOString(),
6371
status,
6472
}
6573

66-
const current = (await scheduledJobRunStorage.getValue()) ?? []
67-
const otherJobRuns = current.filter((r) => r.jobId !== jobId)
68-
const thisJobRuns = current
69-
.filter((r) => r.jobId === jobId)
70-
.sort(
71-
(a, b) =>
72-
new Date(b.startedAt).getTime() - new Date(a.startedAt).getTime(),
73-
)
74-
.slice(0, MAX_RUNS_PER_JOB - 1)
75-
76-
await scheduledJobRunStorage.setValue([
77-
...otherJobRuns,
78-
...thisJobRuns,
79-
jobRun,
80-
])
74+
await putScheduledJobRun(jobRun)
8175
return jobRun
8276
}
8377

78+
// Takes the run rather than its id: the caller already holds it, and merging
79+
// locally avoids re-reading a list to update one row.
8480
const updateJobRun = async (
85-
runId: string,
81+
run: ScheduledJobRun,
8682
updates: Partial<Omit<ScheduledJobRun, 'id' | 'jobId' | 'startedAt'>>,
8783
) => {
88-
const current = (await scheduledJobRunStorage.getValue()) ?? []
89-
await scheduledJobRunStorage.setValue(
90-
current.map((r) => (r.id === runId ? { ...r, ...updates } : r)),
91-
)
84+
await putScheduledJobRun({ ...run, ...updates })
9285
}
9386

87+
// Takes an id, not the job: a snapshot captured before the run would be
88+
// minutes stale by the time this writes, and putting it back would revert any
89+
// edit made while the run was going.
9490
const updateJobLastRunAt = async (jobId: string) => {
95-
const current = (await scheduledJobStorage.getValue()) ?? []
96-
await scheduledJobStorage.setValue(
97-
current.map((j) =>
98-
j.id === jobId ? { ...j, lastRunAt: new Date().toISOString() } : j,
99-
),
100-
)
91+
const jobs = await listScheduledJobsOrNull()
92+
if (jobs === null) return
93+
94+
const updated = applyLastRunAt(jobs, jobId, new Date().toISOString())
95+
if (updated) await putScheduledJob(updated)
10196
}
10297

10398
const executeScheduledJob = async (jobId: string): Promise<void> => {
104-
const job = (await scheduledJobStorage.getValue()).find(
105-
(each) => each.id === jobId,
106-
)
99+
const jobs = await listScheduledJobsOrNull()
100+
if (jobs === null) {
101+
throw new Error('Cannot reach the BrowserOS server to load the job')
102+
}
107103

104+
const job = jobs.find((each) => each.id === jobId)
108105
if (!job) {
109106
throw new Error(`Job not found: ${jobId}`)
110107
}
@@ -120,7 +117,7 @@ export const scheduledJobRuns = async () => {
120117
providerId: job.providerId,
121118
})
122119

123-
await updateJobRun(jobRun.id, {
120+
await updateJobRun(jobRun, {
124121
status: 'completed',
125122
completedAt: new Date().toISOString(),
126123
result: response.text,
@@ -135,7 +132,7 @@ export const scheduledJobRuns = async () => {
135132
: e instanceof Error
136133
? e.message
137134
: String(e)
138-
await updateJobRun(jobRun.id, {
135+
await updateJobRun(jobRun, {
139136
status: 'failed',
140137
completedAt: new Date().toISOString(),
141138
result: errorMessage,
@@ -155,10 +152,11 @@ export const scheduledJobRuns = async () => {
155152
runningMissedJobs = true
156153

157154
try {
158-
const jobs = (await scheduledJobStorage.getValue()).filter(
159-
(j) => j.enabled,
160-
)
161-
const runs = (await scheduledJobRunStorage.getValue()) ?? []
155+
const loadedJobs = await listScheduledJobsOrNull()
156+
const runs = await listScheduledJobRunsOrNull()
157+
if (loadedJobs === null || runs === null) return
158+
159+
const jobs = loadedJobs.filter((j) => j.enabled)
162160
const now = Date.now()
163161
const cutoff = now - TWENTY_FOUR_HOURS_MS
164162

‎packages/browseros-agent/apps/app/lib/schedules/scheduleStorage.ts‎

Lines changed: 6 additions & 143 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,12 @@
11
import { storage } from '@wxt-dev/storage'
2-
import { useEffect, useState } from 'react'
3-
import { sendScheduleMessage } from '@/lib/messaging/schedules/scheduleMessages'
4-
import { createAlarmFromJob } from './createAlarmFromJob'
52
import type { ScheduledJob, ScheduledJobRun } from './scheduleTypes'
63

7-
const getAlarmName = (jobId: string) => `scheduled-job-${jobId}`
4+
/**
5+
* Legacy extension storage for scheduled jobs and their runs.
6+
*
7+
* The server owns both now. These items remain only as the source the one-time
8+
* import reads, and nothing writes them any more.
9+
*/
810

911
export const scheduledJobStorage = storage.defineItem<ScheduledJob[]>(
1012
'local:scheduledJobs',
@@ -19,142 +21,3 @@ export const scheduledJobRunStorage = storage.defineItem<ScheduledJobRun[]>(
1921
fallback: [],
2022
},
2123
)
22-
23-
export const pendingDeletionStorage = storage.defineItem<string[]>(
24-
'local:scheduledJobsPendingDeletion',
25-
{
26-
fallback: [],
27-
},
28-
)
29-
30-
export function useScheduledJobs() {
31-
const [jobs, setJobs] = useState<ScheduledJob[]>([])
32-
33-
useEffect(() => {
34-
scheduledJobStorage.getValue().then(setJobs)
35-
const unwatch = scheduledJobStorage.watch((newValue) => {
36-
setJobs(newValue ?? [])
37-
})
38-
return unwatch
39-
}, [])
40-
41-
const addJob = async (
42-
job: Omit<ScheduledJob, 'id' | 'createdAt' | 'updatedAt'>,
43-
) => {
44-
const now = new Date().toISOString()
45-
const newJob: ScheduledJob = {
46-
id: crypto.randomUUID(),
47-
createdAt: now,
48-
updatedAt: now,
49-
...job,
50-
}
51-
const current = (await scheduledJobStorage.getValue()) ?? []
52-
await scheduledJobStorage.setValue([...current, newJob])
53-
54-
if (newJob.enabled) {
55-
await createAlarmFromJob(newJob)
56-
}
57-
}
58-
59-
const removeJob = async (id: string) => {
60-
await chrome.alarms.clear(getAlarmName(id))
61-
62-
const pending = (await pendingDeletionStorage.getValue()) ?? []
63-
if (!pending.includes(id)) {
64-
await pendingDeletionStorage.setValue([...pending, id])
65-
}
66-
67-
const currentJobs = (await scheduledJobStorage.getValue()) ?? []
68-
await scheduledJobStorage.setValue(currentJobs.filter((j) => j.id !== id))
69-
70-
const currentRuns = (await scheduledJobRunStorage.getValue()) ?? []
71-
await scheduledJobRunStorage.setValue(
72-
currentRuns.filter((r) => r.jobId !== id),
73-
)
74-
}
75-
76-
const toggleJob = async (id: string, enabled: boolean) => {
77-
const current = (await scheduledJobStorage.getValue()) ?? []
78-
const job = current.find((j) => j.id === id)
79-
if (!job) return
80-
81-
const updatedAt = new Date().toISOString()
82-
await scheduledJobStorage.setValue(
83-
current.map((j) => (j.id === id ? { ...j, enabled, updatedAt } : j)),
84-
)
85-
86-
if (enabled) {
87-
await createAlarmFromJob({ ...job, enabled })
88-
} else {
89-
await chrome.alarms.clear(getAlarmName(id))
90-
}
91-
}
92-
93-
const editJob = async (
94-
id: string,
95-
updates: Omit<ScheduledJob, 'id' | 'createdAt' | 'updatedAt'>,
96-
) => {
97-
const current = (await scheduledJobStorage.getValue()) ?? []
98-
const existingJob = current.find((j) => j.id === id)
99-
if (!existingJob) return
100-
101-
const updatedJob: ScheduledJob = {
102-
id,
103-
createdAt: existingJob.createdAt,
104-
updatedAt: new Date().toISOString(),
105-
...updates,
106-
}
107-
await scheduledJobStorage.setValue(
108-
current.map((j) => (j.id === id ? updatedJob : j)),
109-
)
110-
111-
await chrome.alarms.clear(getAlarmName(id))
112-
if (updatedJob.enabled) {
113-
await createAlarmFromJob(updatedJob)
114-
}
115-
}
116-
117-
const runJob = async (id: string) => {
118-
return sendScheduleMessage('runScheduledJob', { jobId: id })
119-
}
120-
121-
return { jobs, addJob, removeJob, editJob, toggleJob, runJob }
122-
}
123-
124-
export function useScheduledJobRuns() {
125-
const [jobRuns, setJobRuns] = useState<ScheduledJobRun[]>([])
126-
127-
useEffect(() => {
128-
scheduledJobRunStorage.getValue().then(setJobRuns)
129-
const unwatch = scheduledJobRunStorage.watch((newValue) => {
130-
setJobRuns(newValue ?? [])
131-
})
132-
return unwatch
133-
}, [])
134-
135-
const addJobRun = async (jobRun: ScheduledJobRun) => {
136-
const current = (await scheduledJobRunStorage.getValue()) ?? []
137-
await scheduledJobRunStorage.setValue([...current, jobRun])
138-
}
139-
140-
const removeJobRun = async (id: string) => {
141-
const current = (await scheduledJobRunStorage.getValue()) ?? []
142-
await scheduledJobRunStorage.setValue(current.filter((r) => r.id !== id))
143-
}
144-
145-
const editJobRun = async (
146-
id: string,
147-
updates: Partial<Omit<ScheduledJobRun, 'id'>>,
148-
) => {
149-
const current = (await scheduledJobRunStorage.getValue()) ?? []
150-
await scheduledJobRunStorage.setValue(
151-
current.map((r) => (r.id === id ? { ...r, ...updates } : r)),
152-
)
153-
}
154-
155-
const cancelJobRun = async (runId: string) => {
156-
return sendScheduleMessage('cancelScheduledJobRun', { runId })
157-
}
158-
159-
return { jobRuns, addJobRun, removeJobRun, editJobRun, cancelJobRun }
160-
}

0 commit comments

Comments
 (0)