feat(supply-chain): 日销任务改为每日 08:00 单次执行,未到数即告警并支持手动补跑
This commit is contained in:
parent
d5443e9f77
commit
49ec0be34f
@ -13,7 +13,7 @@ chmod 600 "$PERSONAL_ENV"
|
||||
mkdir -p "$BASE/shared/logs/marketing-sales-job"
|
||||
cat >"/etc/systemd/system/$SERVICE.service" <<EOF
|
||||
[Unit]
|
||||
Description=Synchronize product archive marketing daily sales with retry
|
||||
Description=Synchronize product archive marketing daily sales once a day
|
||||
After=network-online.target
|
||||
Wants=network-online.target
|
||||
[Service]
|
||||
@ -30,9 +30,9 @@ KillMode=control-group
|
||||
EOF
|
||||
cat >"/etc/systemd/system/$SERVICE.timer" <<EOF
|
||||
[Unit]
|
||||
Description=Marketing daily sales from Beijing 08:00, retry every 10 minutes
|
||||
Description=Marketing daily sales once a day at Beijing 08:00
|
||||
[Timer]
|
||||
OnCalendar=*-*-* 08..23:00/10:00 Asia/Shanghai
|
||||
OnCalendar=*-*-* 08:00:00 Asia/Shanghai
|
||||
OnBootSec=1min
|
||||
AccuracySec=1s
|
||||
Persistent=true
|
||||
|
||||
@ -36,7 +36,7 @@ describe('marketing sales result notifications', () => {
|
||||
expect(buildMarketingSalesNotice(failed)?.content.split('\n\n').map(part => part.split(':')[0])).toEqual([
|
||||
'任务日期', '时间', '原因', '处理',
|
||||
])
|
||||
expect(buildMarketingSalesNotice({ ...completed, phase: 'waiting', lateAt: '2026-10-12T01:30:00.000Z' })?.content).toContain('09:30 尚未读到')
|
||||
expect(buildMarketingSalesNotice({ ...completed, phase: 'waiting', lateAt: '2026-10-12T01:30:00.000Z' })?.content).toContain('08:00 未读到')
|
||||
expect(buildMarketingSalesNotice({ ...failed, phase: 'completed' })?.kind).toBe('recovered')
|
||||
})
|
||||
|
||||
|
||||
@ -99,34 +99,35 @@ describe('daily completion and recovery', () => {
|
||||
expect(f.deps.lock.exclusive).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('keeps yesterday data and retries at 08:10 even if the first query ends at 08:00:04', async () => {
|
||||
it('keeps yesterday data at 08:00, alerts once and succeeds on a later manual run', async () => {
|
||||
const f = fixture()
|
||||
f.source.readSnapshot.mockResolvedValueOnce(buildMarketingSalesSnapshot([sample({ sourceDate: '2026-10-11' })]))
|
||||
const result = await f.service.run()
|
||||
expect(result.status).toBe('WAITING_SOURCE')
|
||||
expect(result.state?.nextCheckAt).toBe('2026-10-12T00:10:00.000Z')
|
||||
expect(result.state).toEqual(expect.objectContaining({ lateAt: '2026-10-12T00:00:04.000Z', nextCheckAt: '2026-10-12T00:00:04.000Z' }))
|
||||
expect(f.repository.applySnapshot).not.toHaveBeenCalled()
|
||||
// 自动链路每日仅 08:00 触发一次;人工补跑(手动 start 服务)随时可行。
|
||||
f.setClock('2026-10-12T08:10:00+08:00')
|
||||
expect((await f.service.run()).status).toBe('APPLIED')
|
||||
expect(f.source.readSnapshot).toHaveBeenCalledTimes(2)
|
||||
expect(f.deps.journal.recordCheck).toHaveBeenCalledWith(expect.objectContaining({ available: true }))
|
||||
})
|
||||
|
||||
it('continues waiting after 09:30 and marks the overdue condition', async () => {
|
||||
it('marks the overdue condition immediately when the 08:00 source is not ready', async () => {
|
||||
const f = fixture()
|
||||
f.setClock('2026-10-12T09:30:00+08:00')
|
||||
f.source.readSnapshot.mockResolvedValue(buildMarketingSalesSnapshot([sample({ sourceDate: '2026-10-11' })]))
|
||||
const result = await f.service.run()
|
||||
expect(result.state).toEqual(expect.objectContaining({ phase: 'waiting', lateAt: '2026-10-12T01:30:00.000Z', nextCheckAt: '2026-10-12T01:40:00.000Z' }))
|
||||
expect(result.state).toEqual(expect.objectContaining({ phase: 'waiting', lateAt: '2026-10-12T01:30:00.000Z', nextCheckAt: '2026-10-12T01:30:00.000Z' }))
|
||||
expect(f.repository.applySnapshot).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('records a safe failure and retries successfully ten minutes later', async () => {
|
||||
it('records a safe failure and succeeds on a manual rerun', async () => {
|
||||
const f = fixture()
|
||||
f.source.readSnapshot.mockRejectedValueOnce(Object.assign(new Error('unsafe connection text'), { code: 'ETIMEOUT' }))
|
||||
await expect(f.service.run()).rejects.toThrow('unsafe connection text')
|
||||
expect(f.source.close).toHaveBeenCalledTimes(1)
|
||||
expect(f.repository.recordFailure).toHaveBeenCalledWith(expect.objectContaining({ phase: 'retry', firstErrorCode: 'ETIMEOUT', nextCheckAt: '2026-10-12T00:10:00.000Z' }), null, 'ETIMEOUT')
|
||||
expect(f.repository.recordFailure).toHaveBeenCalledWith(expect.objectContaining({ phase: 'retry', firstErrorCode: 'ETIMEOUT', nextCheckAt: '2026-10-12T00:00:04.000Z' }), null, 'ETIMEOUT')
|
||||
f.setClock('2026-10-12T08:10:00+08:00')
|
||||
const result = await f.service.run()
|
||||
expect(result.state).toEqual(expect.objectContaining({ phase: 'completed', attemptCount: 2, firstErrorCode: 'ETIMEOUT' }))
|
||||
|
||||
@ -26,8 +26,8 @@ export function buildMarketingSalesNotice(state: MarketingSalesState) {
|
||||
'结果:当天数据已同步完成,当天停止查询源表。',
|
||||
] : [
|
||||
`时间:${beijingTime(state.failureAt || state.lateAt)}`,
|
||||
`原因:${state.failureAt ? state.firstErrorCode || 'MARKETING_SALES_SYNC_FAILED' : '09:30 尚未读到当天完整数据'}`,
|
||||
'处理:每 10 分钟自动补跑,成功后另行通知。',
|
||||
`原因:${state.failureAt ? state.firstErrorCode || 'MARKETING_SALES_SYNC_FAILED' : '08:00 未读到当天完整数据'}`,
|
||||
'处理:该任务每日 08:00 自动执行一次,今日已停止自动重试,请排查后手动补跑。',
|
||||
]),
|
||||
// DingTalk renders the center's text messages as Markdown; blank lines keep fields separate.
|
||||
].join('\n\n')
|
||||
|
||||
@ -25,8 +25,9 @@ export type MarketingSalesState = {
|
||||
|
||||
const HOUR = 3_600_000
|
||||
export const MARKETING_SALES_DAILY_MINUTE = 8 * 60
|
||||
export const MARKETING_SALES_RETRY_MINUTES = 10
|
||||
export const MARKETING_SALES_LATE_MINUTE = 9 * 60 + 30
|
||||
// 2026-10-14 用户确认上游日销数据稳定在 08:00 前就绪:每日 08:00 仅自动执行一次。
|
||||
// 未到数即推送告警;"自动每天一次"由 systemd timer 的 OnCalendar 控制,
|
||||
// 这里保持 nextCheckAt=now,人工触发补跑(手动 start 服务)随时可行。
|
||||
export const INITIAL_MARKETING_SALES_STATE: MarketingSalesState = {
|
||||
mode: 'daily', dailyMinute: MARKETING_SALES_DAILY_MINUTE, observations: [],
|
||||
}
|
||||
@ -45,8 +46,7 @@ export function marketingSalesCheckDue(state: MarketingSalesState, now: Date) {
|
||||
export function resetMarketingSalesSchedule(state: MarketingSalesState, now: Date): MarketingSalesState {
|
||||
return { ...state, mode: 'daily', dailyMinute: MARKETING_SALES_DAILY_MINUTE,
|
||||
runDate: beijingDate(now), phase: 'retry', lastCheckedAt: now.toISOString(),
|
||||
nextCheckAt: new Date((Math.floor(now.getTime() / (MARKETING_SALES_RETRY_MINUTES * 60_000)) + 1)
|
||||
* MARKETING_SALES_RETRY_MINUTES * 60_000).toISOString() }
|
||||
nextCheckAt: now.toISOString() }
|
||||
}
|
||||
|
||||
export function observeMarketingSales(state: MarketingSalesState, sourceDate: string, fingerprint: string, now: Date): MarketingSalesState {
|
||||
@ -75,10 +75,10 @@ export function startMarketingSalesAttempt(state: MarketingSalesState, now: Date
|
||||
}
|
||||
|
||||
export function waitForMarketingSales(state: MarketingSalesState, now: Date): MarketingSalesState {
|
||||
// 08:00 单次执行:首查未到数即视为迟到并推送告警;当天不再自动补查。
|
||||
return { ...resetMarketingSalesSchedule(state, now), phase: 'waiting',
|
||||
lastNotAvailableAt: now.toISOString(),
|
||||
...(beijingMinute(now) >= MARKETING_SALES_LATE_MINUTE
|
||||
? { lateAt: state.lateAt || now.toISOString() } : {}) }
|
||||
lateAt: state.lateAt || now.toISOString() }
|
||||
}
|
||||
|
||||
export function marketingSalesErrorCode(error: unknown) {
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user