speatzle/morffix
1
0
Fork 0
Code Issues Pull requests Projects Releases 2 Packages Wiki Activity Actions

fix task assign error checking
All checks were successful
/ release (push) Successful in 40s
Details

Browse source
This commit is contained in:
Samuel Lorch 2024-07-06 16:50:39 +02:00
parent 8046093e9e
commit 2dec6a6271
1 changed files with 22 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

37
server/task.go
View file

@ -390,24 +390,31 @@ func assignQueuedTasks(ctx context.Context) error {
} }
_, err = rpcServer.Call(ctx, Workers[i].Conn, "task-start", taskStart, nil) _, err = rpcServer.Call(ctx, Workers[i].Conn, "task-start", taskStart, nil)
if strings.HasSuffix(err.Error(), constants.ErrTaskIsAlreadyRunning.Error()) { if err != nil {
// Task was started previously but something went wrong and we are out of sync if strings.HasSuffix(err.Error(), constants.ErrTaskIsAlreadyRunning.Error()) {
slog.WarnContext(ctx, "Task is apparently already Running on this Worker, thats bad", "task_id", taskStart.ID, "worker", Workers[i].Name) // Task was started previously but something went wrong and we are out of sync
slog.WarnContext(ctx, "Task is apparently already Running on this Worker, thats bad", "task_id", taskStart.ID, "worker", Workers[i].Name)
_, err = db.Exec(ctx, "UPDATE tasks SET status = $2, log = log || $3 WHERE id = $1", taskStart.ID, constants.TASK_STATUS_ASSIGNED, []string{fmt.Sprintf("%v MASTER: Task Start, Task Already Running!", time.Now())}) _, err = db.Exec(ctx, "UPDATE tasks SET status = $2, log = log || $3 WHERE id = $1", taskStart.ID, constants.TASK_STATUS_ASSIGNED, []string{fmt.Sprintf("%v MASTER: Task Start, Task Already Running!", time.Now())})
if err != nil { if err != nil {
return fmt.Errorf("Updating Task status during already running error: %w", err) return fmt.Errorf("Updating Task status during already running error: %w", err)
} }
} else if errors.Is(err, constants.ErrRPCRequestTimeout) { } else if errors.Is(err, constants.ErrRPCRequestTimeout) {
// We really don't know whats going on, might be slow response, oom, disk full or a bug // We really don't know whats going on, might be slow response, oom, disk full or a bug
slog.WarnContext(ctx, "Task start Timed Out", "task_id", taskStart.ID, "worker", Workers[i].Name) slog.WarnContext(ctx, "Task start Timed Out", "task_id", taskStart.ID, "worker", Workers[i].Name)
_, err = db.Exec(ctx, "UPDATE tasks SET status = $2, log = log || $3 WHERE id = $1", taskStart.ID, constants.TASK_STATUS_UNKNOWN, []string{fmt.Sprintf("%v MASTER: Task Start RPC Call Timed Out!", time.Now())}) _, err = db.Exec(ctx, "UPDATE tasks SET status = $2, log = log || $3 WHERE id = $1", taskStart.ID, constants.TASK_STATUS_UNKNOWN, []string{fmt.Sprintf("%v MASTER: Task Start RPC Call Timed Out!", time.Now())})
if err != nil { if err != nil {
return fmt.Errorf("Updating Unknown Task Status due to Timeout while starting Task: %w", err) return fmt.Errorf("Updating Unknown Task Status due to Timeout while starting Task: %w", err)
}
} else {
slog.ErrorContext(ctx, "Task start Error", "task_id", taskStart.ID, "worker", Workers[i].Name)
_, err = db.Exec(ctx, "UPDATE tasks SET status = $2, log = log || $3 WHERE id = $1", taskStart.ID, constants.TASK_STATUS_UNKNOWN, []string{fmt.Sprintf("%v MASTER: Task Start Error: %v", time.Now(), err.Error())})
if err != nil {
return fmt.Errorf("Updating Unknown Task Status due to Error while starting Task: %w", err)
}
} }
} else if err != nil {
return fmt.Errorf("Error Starting Task: %w", err)
} }
err = tx.Commit(ctx) err = tx.Commit(ctx)