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

use known task id instead of response

Browse source
This commit is contained in:
speatzle 2024-05-21 16:56:02 +02:00
parent 3dfb113aaf
commit 290acc158a
1 changed files with 7 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
server/worker.go
View file

@ -194,6 +194,7 @@ func updateWorkerTaskStatus(ctx context.Context) {
for _, taskStatusRequest := range taskStatusRequests { for _, taskStatusRequest := range taskStatusRequests {
tsr := taskStatusRequest tsr := taskStatusRequest
taskID := tsr.ID
wg.Add(1) wg.Add(1)
go func() { go func() {
@ -204,19 +205,19 @@ func updateWorkerTaskStatus(ctx context.Context) {
// Find better way to compare errors which where send via websocket // Find better way to compare errors which where send via websocket
if strings.HasSuffix(err.Error(), constants.ErrTaskDoesNotExist.Error()) { if strings.HasSuffix(err.Error(), constants.ErrTaskDoesNotExist.Error()) {
// Worker says it does not know of this task, mark it failed so that we don't asks the worker about it again and again // Worker says it does not know of this task, mark it failed so that we don't asks the worker about it again and again
slog.ErrorContext(ctx, "Task is unknown by worker, Failing...", "err", err, "id", ts.Task.ID) slog.ErrorContext(ctx, "Task is unknown by worker, Failing...", "err", err, "id", taskID)
_, err = db.Exec(ctx, "UPDATE tasks SET status = $2, log = log || $3 WHERE id = $1", ts.Task.ID, constants.TASK_STATUS_FAILED, []string{"Task Failed because it is unknown to Assigned Worker"}) _, err = db.Exec(ctx, "UPDATE tasks SET status = $2, log = log || $3 WHERE id = $1", taskID, constants.TASK_STATUS_FAILED, []string{"Task Failed because it is unknown to Assigned Worker"})
if err != nil { if err != nil {
slog.ErrorContext(ctx, "Updating Failed Task Status", "err", err) slog.ErrorContext(ctx, "Updating Failed Task Status", "err", err)
return return
} }
slog.Info("Updating task done", "id", ts.Task.ID, "status", constants.TASK_STATUS_FAILED) slog.Info("Updating task done", "id", taskID, "status", constants.TASK_STATUS_FAILED)
return return
} else if err != nil { } else if err != nil {
slog.ErrorContext(ctx, "Getting Task Status", "err", err) slog.ErrorContext(ctx, "Getting Task Status", "err", err)
_, err = db.Exec(ctx, "UPDATE tasks SET status = $2 WHERE id = $1", ts.Task.ID, constants.TASK_STATUS_UNKNOWN) _, err = db.Exec(ctx, "UPDATE tasks SET status = $2 WHERE id = $1", taskID, constants.TASK_STATUS_UNKNOWN)
if err != nil { if err != nil {
slog.ErrorContext(ctx, "Updating Unknown Task Status", "err", err) slog.ErrorContext(ctx, "Updating Unknown Task Status", "err", err)
return return
@ -224,7 +225,7 @@ func updateWorkerTaskStatus(ctx context.Context) {
return return
} }
_, err = db.Exec(ctx, "UPDATE tasks SET status = $2, log = log || $3 WHERE id = $1", ts.Task.ID, ts.Task.Status, ts.Task.Log) _, err = db.Exec(ctx, "UPDATE tasks SET status = $2, log = log || $3 WHERE id = $1", taskID, ts.Task.Status, ts.Task.Log)
if err != nil { if err != nil {
slog.ErrorContext(ctx, "Updating Task Status", "err", err) slog.ErrorContext(ctx, "Updating Task Status", "err", err)
return return
@ -247,7 +248,7 @@ func updateWorkerTaskStatus(ctx context.Context) {
// Tell Worker to Delete Finished Tasks // Tell Worker to Delete Finished Tasks
if ts.Task.Status == constants.TASK_STATUS_FAILED || ts.Task.Status == constants.TASK_STATUS_SUCCESS { if ts.Task.Status == constants.TASK_STATUS_FAILED || ts.Task.Status == constants.TASK_STATUS_SUCCESS {
_, err := rpcServer.Call(ctx, w.Conn, "task-delete", ts.Task.ID, nil) _, err := rpcServer.Call(ctx, w.Conn, "task-delete", taskID, nil)
if err != nil { if err != nil {
slog.ErrorContext(ctx, "Deleting Finished Task From Worker", "err", err) slog.ErrorContext(ctx, "Deleting Finished Task From Worker", "err", err)
return return