mirror of
https://github.com/Rain-kl/OpenFlare.git
synced 2026-10-02 14:56:38 +08:00
fix(drivers): propagate app-lifetime context through inproc cron/worker drivers
cron 触发与 worker 执行的任务现在继承应用生命周期 context(关闭时级联取消,带超时子上下文), 替代裸 context.Background()。contextcheck 清零。 lint_issues 26→24
This commit is contained in:
@@ -120,7 +120,7 @@ func NewMockStorageService() *MockStorageService {
|
||||
}
|
||||
|
||||
// Put uploads an object into mock storage.
|
||||
func (m *MockStorageService) Put(_ context.Context, key string, body io.Reader, _ int64, contentType string) (contracts.StoragePutResult, error) {
|
||||
func (m *MockStorageService) Put(_ context.Context, key string, body io.Reader, _ int64, _ string) (contracts.StoragePutResult, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
data, err := io.ReadAll(body)
|
||||
|
||||
@@ -56,8 +56,9 @@ func (p *Plugin) Type() core.DriverType {
|
||||
return core.DriverTypeScheduler
|
||||
}
|
||||
|
||||
// Start boots the in-process cron scheduler.
|
||||
func (p *Plugin) Start(_ context.Context) error {
|
||||
// Start boots the in-process cron scheduler. ctx is the app-lifetime context;
|
||||
// cron-dispatched tasks carry it so cancellation propagates on shutdown.
|
||||
func (p *Plugin) Start(ctx context.Context) error {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
|
||||
@@ -66,7 +67,7 @@ func (p *Plugin) Start(_ context.Context) error {
|
||||
p.scheduler = newInprocScheduler(p.coreCtx.Schedules(), p.coreCtx.Tasks(), taskSvc)
|
||||
}
|
||||
|
||||
return p.scheduler.Start()
|
||||
return p.scheduler.Start(ctx)
|
||||
}
|
||||
|
||||
// Stop terminates the in-process cron scheduler.
|
||||
|
||||
@@ -36,7 +36,11 @@ func newInprocScheduler(scheduleReg extpoints.ScheduleExtension, taskReg extpoin
|
||||
}
|
||||
}
|
||||
|
||||
func (s *inprocScheduler) Start() error {
|
||||
func (s *inprocScheduler) Start(ctx context.Context) error {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
@@ -46,7 +50,7 @@ func (s *inprocScheduler) Start() error {
|
||||
|
||||
if s.scheduleReg != nil {
|
||||
for _, def := range s.scheduleReg.Schedules() {
|
||||
s.registerJob(def)
|
||||
s.registerJob(ctx, def)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -70,7 +74,7 @@ func (s *inprocScheduler) Stop() {
|
||||
|
||||
const standardCronFields = 5
|
||||
|
||||
func (s *inprocScheduler) registerJob(def extpoints.ScheduleDefinition) {
|
||||
func (s *inprocScheduler) registerJob(ctx context.Context, def extpoints.ScheduleDefinition) {
|
||||
spec := def.Spec
|
||||
taskType := def.TaskType
|
||||
|
||||
@@ -94,8 +98,8 @@ func (s *inprocScheduler) registerJob(def extpoints.ScheduleDefinition) {
|
||||
|
||||
_, err := s.cronRunner.AddFunc(cronSpec, func() {
|
||||
if s.taskSvc != nil {
|
||||
if _, dispatchErr := s.taskSvc.Dispatch(context.Background(), taskType, payloadBytes, "inproc_cron"); dispatchErr != nil {
|
||||
logger.ErrorF(context.Background(), "driver_inproc_cron: dispatch task %q failed: %v", taskType, dispatchErr)
|
||||
if _, dispatchErr := s.taskSvc.Dispatch(ctx, taskType, payloadBytes, "inproc_cron"); dispatchErr != nil {
|
||||
logger.ErrorF(ctx, "driver_inproc_cron: dispatch task %q failed: %v", taskType, dispatchErr)
|
||||
}
|
||||
return
|
||||
}
|
||||
@@ -107,15 +111,15 @@ func (s *inprocScheduler) registerJob(def extpoints.ScheduleDefinition) {
|
||||
if timeout <= 0 {
|
||||
timeout = 5 * time.Minute
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
||||
runCtx, cancel := context.WithTimeout(ctx, timeout)
|
||||
defer cancel()
|
||||
_ = invokeHandler(ctx, td.Handler, payloadBytes)
|
||||
_ = invokeHandler(runCtx, td.Handler, payloadBytes)
|
||||
})
|
||||
}
|
||||
}
|
||||
})
|
||||
if err != nil {
|
||||
logger.ErrorF(context.Background(), "driver_inproc_cron: invalid cron spec %q for task %q: %v", spec, taskType, err)
|
||||
logger.ErrorF(ctx, "driver_inproc_cron: invalid cron spec %q for task %q: %v", spec, taskType, err)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -35,6 +35,10 @@ type InprocQueue struct {
|
||||
running atomic.Bool
|
||||
stopCh chan struct{}
|
||||
wg sync.WaitGroup
|
||||
|
||||
// baseCtx is the app-lifetime context captured at Start; task handlers
|
||||
// derive their timeouts from it so shutdown cancellation propagates.
|
||||
baseCtx context.Context
|
||||
}
|
||||
|
||||
// NewInprocQueue creates a new InprocQueue with a given concurrency and queue capacity.
|
||||
@@ -83,12 +87,16 @@ func (q *InprocQueue) Enqueue(taskType string, payload []byte, source string) (s
|
||||
}
|
||||
}
|
||||
|
||||
// Start begins processing tasks with the worker pool.
|
||||
func (q *InprocQueue) Start() {
|
||||
// Start begins processing tasks with the worker pool. ctx is the app-lifetime
|
||||
// context used as the parent for per-task execution contexts.
|
||||
func (q *InprocQueue) Start(ctx context.Context) {
|
||||
if !q.running.CompareAndSwap(false, true) {
|
||||
return
|
||||
}
|
||||
|
||||
if q.baseCtx == nil {
|
||||
q.baseCtx = ctx
|
||||
}
|
||||
for i := 0; i < q.concurrency; i++ {
|
||||
q.wg.Add(1)
|
||||
util.Go(func() {
|
||||
@@ -149,7 +157,7 @@ func (q *InprocQueue) executeTask(msg TaskMessage) {
|
||||
timeout = 5 * time.Minute
|
||||
}
|
||||
|
||||
taskCtx, cancel := context.WithTimeout(context.Background(), timeout)
|
||||
taskCtx, cancel := context.WithTimeout(q.baseCtx, timeout)
|
||||
defer cancel()
|
||||
|
||||
err := invokeHandler(taskCtx, td.Handler, msg.Payload)
|
||||
|
||||
@@ -119,8 +119,9 @@ func (p *Plugin) Type() core.DriverType {
|
||||
return core.DriverTypeWorker
|
||||
}
|
||||
|
||||
// Start initiates task consumption.
|
||||
func (p *Plugin) Start(_ context.Context) error {
|
||||
// Start initiates task consumption. ctx is the app-lifetime context handed to
|
||||
// task executions so shutdown cancellation propagates.
|
||||
func (p *Plugin) Start(ctx context.Context) error {
|
||||
if p.queue == nil {
|
||||
p.queue = NewInprocQueue(p.concurrency, p.queueCapacity, p.coreCtx.Tasks())
|
||||
}
|
||||
@@ -129,7 +130,7 @@ func (p *Plugin) Start(_ context.Context) error {
|
||||
globalQueue = p.queue
|
||||
globalMu.Unlock()
|
||||
|
||||
p.queue.Start()
|
||||
p.queue.Start(ctx)
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user