Skip to content

Commit 57bea95

Browse files
ulemonsulemons
authored
feat: move to continue as new pattern (CM-1277) (#4262)
Signed-off-by: Umberto Sgueglia <usgueglia@contractor.linuxfoundation.org>
1 parent a17f2fc commit 57bea95

6 files changed

Lines changed: 54 additions & 94 deletions

File tree

‎services/apps/packages_worker/src/activities.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ export {
1212
} from './npm/activities'
1313
export * from './deps-dev/activities'
1414
export { osvSyncEcosystem, osvDeriveCriticalFlag } from './osv/activities'
15-
export { processMavenCriticalBatch, processMavenNonCriticalBatch } from './maven/activities'
15+
export { processMavenCriticalBatch } from './maven/activities'
1616
export { criticalityComputePageRank, rankPackages } from './criticality/activities'
1717
export {
1818
cargoDownloadAndLoad,

‎services/apps/packages_worker/src/bin/maven-worker.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,8 @@
1-
import { scheduleMavenCritical } from '../maven/schedule'
1+
import { scheduleMavenIngestion } from '../maven/schedule'
22
import { svc } from '../service'
33

44
setImmediate(async () => {
55
await svc.init()
6-
await scheduleMavenCritical()
6+
await scheduleMavenIngestion()
77
await svc.start()
88
})

‎services/apps/packages_worker/src/maven/activities.ts‎

Lines changed: 0 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -19,20 +19,3 @@ export async function processMavenCriticalBatch(): Promise<BatchResult> {
1919
log.info({ ...result }, 'Maven critical batch complete')
2020
return result
2121
}
22-
23-
export async function processMavenNonCriticalBatch(): Promise<BatchResult> {
24-
const config = getMavenConfig()
25-
const qx = await getPackagesDb()
26-
// Non-critical is DB-only (no POM fetch); the flag is unused on this path.
27-
const result = await processBatch(qx, config, false, false)
28-
log.info(
29-
{
30-
processed: result.processed,
31-
skipped: result.skipped,
32-
unchanged: result.unchanged,
33-
error: result.error,
34-
},
35-
'Maven non-critical batch complete',
36-
)
37-
return result
38-
}

‎services/apps/packages_worker/src/maven/schedule.ts‎

Lines changed: 36 additions & 61 deletions
Original file line numberDiff line numberDiff line change
@@ -1,78 +1,53 @@
11
import { ScheduleAlreadyRunning, ScheduleOverlapPolicy } from '@temporalio/client'
22

33
import { svc } from '../service'
4-
import { mavenCriticalWorkflow } from '../workflows'
4+
import { ingestMavenPackages } from '../workflows'
55

6-
export async function scheduleMavenCritical(): Promise<void> {
6+
const LEGACY_SCHEDULE_ID = 'maven-critical'
7+
8+
export async function scheduleMavenIngestion(): Promise<void> {
79
const { temporal } = svc
810
if (!temporal) throw new Error('Temporal client not initialized')
911

10-
const scheduleOptions: Parameters<typeof temporal.schedule.create>[0] = {
11-
scheduleId: 'maven-critical',
12-
spec: {
13-
cronExpressions: ['*/1 * * * *'],
14-
},
15-
policies: {
16-
overlap: ScheduleOverlapPolicy.SKIP,
17-
catchupWindow: '1 hour',
18-
},
19-
action: {
20-
type: 'startWorkflow',
21-
workflowType: mavenCriticalWorkflow,
22-
taskQueue: 'packages-worker',
23-
workflowExecutionTimeout: '15 minutes',
24-
retry: {
25-
initialInterval: '30 seconds',
26-
backoffCoefficient: 2,
27-
maximumAttempts: 3,
28-
},
29-
args: [],
30-
},
12+
try {
13+
await temporal.schedule.getHandle(LEGACY_SCHEDULE_ID).delete()
14+
svc.log.info({ scheduleId: LEGACY_SCHEDULE_ID }, 'Deleted legacy schedule.')
15+
} catch (err) {
16+
svc.log.warn(
17+
{ err, scheduleId: LEGACY_SCHEDULE_ID },
18+
'Failed to delete legacy schedule (may not exist).',
19+
)
3120
}
3221

3322
try {
34-
await temporal.schedule.create(scheduleOptions)
23+
await temporal.schedule.create({
24+
scheduleId: 'maven-ingestion',
25+
spec: {
26+
cronExpressions: ['0 9 * * *'],
27+
},
28+
policies: {
29+
overlap: ScheduleOverlapPolicy.SKIP,
30+
catchupWindow: '1 hour',
31+
},
32+
action: {
33+
type: 'startWorkflow',
34+
workflowType: ingestMavenPackages,
35+
workflowId: 'maven-daily-enrichment',
36+
taskQueue: 'packages-worker',
37+
workflowRunTimeout: '24 hours',
38+
retry: {
39+
initialInterval: '30 seconds',
40+
backoffCoefficient: 2,
41+
maximumAttempts: 3,
42+
},
43+
args: [],
44+
},
45+
})
3546
} catch (err) {
3647
if (err instanceof ScheduleAlreadyRunning) {
37-
svc.log.info('Schedule maven-critical already exists, skipping creation.')
48+
svc.log.info('Schedule maven-ingestion already exists, skipping creation.')
3849
} else {
3950
throw err
4051
}
4152
}
4253
}
43-
44-
// export async function scheduleMavenNonCritical(): Promise<void> {
45-
// const { temporal } = svc
46-
// if (!temporal) throw new Error('Temporal client not initialized')
47-
48-
// try {
49-
// await temporal.schedule.create({
50-
// scheduleId: 'maven-non-critical',
51-
// spec: {
52-
// cronExpressions: ['*/10 * * * *'],
53-
// },
54-
// policies: {
55-
// overlap: ScheduleOverlapPolicy.SKIP,
56-
// catchupWindow: '1 hour',
57-
// },
58-
// action: {
59-
// type: 'startWorkflow',
60-
// workflowType: mavenNonCriticalWorkflow,
61-
// taskQueue: 'packages-worker',
62-
// workflowExecutionTimeout: '5 minutes',
63-
// retry: {
64-
// initialInterval: '30 seconds',
65-
// backoffCoefficient: 2,
66-
// maximumAttempts: 3,
67-
// },
68-
// args: [],
69-
// },
70-
// })
71-
// } catch (err) {
72-
// if (err instanceof ScheduleAlreadyRunning) {
73-
// svc.log.info('Schedule maven-non-critical already registered.')
74-
// } else {
75-
// throw err
76-
// }
77-
// }
78-
// }

‎services/apps/packages_worker/src/maven/workflows.ts‎

Lines changed: 14 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -1,19 +1,21 @@
1-
import { proxyActivities } from '@temporalio/workflow'
1+
import { continueAsNew, log, proxyActivities } from '@temporalio/workflow'
22

33
import type * as activities from './activities'
44

5-
const { processMavenCriticalBatch } = proxyActivities<typeof activities>({
5+
const acts = proxyActivities<typeof activities>({
66
startToCloseTimeout: '15 minutes',
7+
retry: {
8+
initialInterval: '30 seconds',
9+
backoffCoefficient: 2,
10+
maximumAttempts: 5,
11+
},
712
})
813

9-
const { processMavenNonCriticalBatch } = proxyActivities<typeof activities>({
10-
startToCloseTimeout: '5 minutes',
11-
})
12-
13-
export async function mavenCriticalWorkflow(): Promise<void> {
14-
await processMavenCriticalBatch()
15-
}
16-
17-
export async function mavenNonCriticalWorkflow(): Promise<void> {
18-
await processMavenNonCriticalBatch()
14+
export async function ingestMavenPackages(): Promise<void> {
15+
const result = await acts.processMavenCriticalBatch()
16+
if (result.processed + result.skipped + result.unchanged === 0) {
17+
log.info('Maven ingestion complete — no more work, exiting.', { ...result })
18+
return
19+
}
20+
await continueAsNew<typeof ingestMavenPackages>()
1921
}

‎services/apps/packages_worker/src/workflows/index.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@ export {
1515
ingestDependentCounts,
1616
} from '../deps-dev/workflows'
1717
export { osvSync } from '../osv/workflows'
18-
export { mavenCriticalWorkflow, mavenNonCriticalWorkflow } from '../maven/workflows'
18+
export { ingestMavenPackages } from '../maven/workflows'
1919
export { ingestScorecard } from '../scorecard/workflows'
2020
export { rankPackagesWorkflow } from '../criticality/workflow'
2121
export { cargoSyncWorkflow } from '../cargo/workflows'

0 commit comments

Comments
 (0)