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

Don't shadow error
All checks were successful
/ release (push) Successful in 39s
Details

Browse source
This commit is contained in:
Samuel Lorch 2025-06-01 00:43:20 +02:00
parent de06175b8d
commit 7ef82b1827
3 changed files with 43 additions and 43 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

36
server/task.go
View file

@ -436,39 +436,39 @@ func assignQueuedTasks(ctx context.Context) error {
// Task was started previously but something went wrong and we are out of sync // 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) slog.WarnContext(ctx, "Task is apparently already Running on this Worker, thats bad", "task_id", taskStart.ID, "worker", Workers[i].Name)
_, err = tx.Exec(ctx, "UPDATE tasks SET status = $2 WHERE id = $1", taskStart.ID, constants.TASK_STATUS_ASSIGNED) _, err2 := tx.Exec(ctx, "UPDATE tasks SET status = $2 WHERE id = $1", taskStart.ID, constants.TASK_STATUS_ASSIGNED)
if err != nil { if err2 != nil {
return fmt.Errorf("Updating Task status during already running error: %w", err) return fmt.Errorf("Updating Task status during already running error: %w", err2)
} }
_, err = tx.Exec(ctx, "INSERT INTO logs(task_id, message) VALUES($1,$2)", taskStart.ID, fmt.Sprintf("%v MASTER: Task Start, Task Already Running!", time.Now())) _, err2 = tx.Exec(ctx, "INSERT INTO logs(task_id, message) VALUES($1,$2)", taskStart.ID, fmt.Sprintf("%v MASTER: Task Start, Task Already Running!", time.Now()))
if err != nil { if err2 != nil {
return fmt.Errorf("Updating Task log during already running error: %w", err) return fmt.Errorf("Updating Task log during already running error: %w", err2)
} }
} 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 = tx.Exec(ctx, "UPDATE tasks SET status = $2 WHERE id = $1", taskStart.ID, constants.TASK_STATUS_UNKNOWN) _, err2 := tx.Exec(ctx, "UPDATE tasks SET status = $2 WHERE id = $1", taskStart.ID, constants.TASK_STATUS_UNKNOWN)
if err != nil { if err2 != 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", err2)
} }
_, err = tx.Exec(ctx, "INSERT INTO logs(task_id, message) VALUES($1,$2)", taskStart.ID, fmt.Sprintf("%v MASTER: Task Start RPC Call Timed Out!", time.Now())) _, err2 = tx.Exec(ctx, "INSERT INTO logs(task_id, message) VALUES($1,$2)", taskStart.ID, fmt.Sprintf("%v MASTER: Task Start RPC Call Timed Out!", time.Now()))
if err != nil { if err2 != nil {
return fmt.Errorf("Updating Unknown Task log due to Timeout while starting Task: %w", err) return fmt.Errorf("Updating Unknown Task log due to Timeout while starting Task: %w", err2)
} }
} else { } else {
slog.ErrorContext(ctx, "Task start Error", "task_id", taskStart.ID, "worker", Workers[i].Name) slog.ErrorContext(ctx, "Task start Error", "task_id", taskStart.ID, "worker", Workers[i].Name)
_, err = tx.Exec(ctx, "UPDATE tasks SET status = $2 WHERE id = $1", taskStart.ID, constants.TASK_STATUS_UNKNOWN) _, err2 := tx.Exec(ctx, "UPDATE tasks SET status = $2 WHERE id = $1", taskStart.ID, constants.TASK_STATUS_UNKNOWN)
if err != nil { if err2 != nil {
return fmt.Errorf("Updating Unknown Task Status due to Error while starting Task: %w", err) return fmt.Errorf("Updating Unknown Task Status due to Error while starting Task: %w", err2)
} }
_, err = tx.Exec(ctx, "INSERT INTO logs(task_id, message) VALUES($1,$2)", taskStart.ID, fmt.Sprintf("%v MASTER: Task Start Error: %v", time.Now(), err.Error())) _, err2 = tx.Exec(ctx, "INSERT INTO logs(task_id, message) VALUES($1,$2)", taskStart.ID, fmt.Sprintf("%v MASTER: Task Start Error: %v", time.Now(), err.Error()))
if err != nil { if err2 != nil {
return fmt.Errorf("Updating Unknown Task Log due to Error while starting Task: %w", err) return fmt.Errorf("Updating Unknown Task Log due to Error while starting Task: %w", err2)
} }
} }
} }

2
server/upload.go
View file

@ -214,7 +214,7 @@ func errorUpload(r *http.Request, w http.ResponseWriter, taskID int, msg string,
http.Error(w, msg+": "+err.Error(), http.StatusInternalServerError) http.Error(w, msg+": "+err.Error(), http.StatusInternalServerError)
if taskID != 0 { if taskID != 0 {
_, err2 := db.Exec(context.TODO(), "INSERT INTO logs(task_id, message) VALUES($1,$2)", taskID, fmt.Sprintf("%v MASTER: upload error: "+msg+": "+err.Error(), time.Now())) _, err2 := db.Exec(context.TODO(), "INSERT INTO logs(task_id, message) VALUES($1,$2)", taskID, fmt.Sprintf("%v MASTER: upload error: "+msg+": "+err.Error(), time.Now()))
if err != nil { if err2 != nil {
slog.ErrorContext(r.Context(), "Updating task log with upload error", "err", err2) slog.ErrorContext(r.Context(), "Updating task log with upload error", "err", err2)
} }
} }

48
server/worker.go
View file

@ -205,9 +205,9 @@ func updateWorkerTaskStatus(ctx context.Context) {
_, err := rpcServer.Call(ctx, w.Conn, "task-status", tsr, &ts) _, err := rpcServer.Call(ctx, w.Conn, "task-status", tsr, &ts)
if err != nil { if err != nil {
tx, err := db.Begin(ctx) tx, err2 := db.Begin(ctx)
if err != nil { if err2 != nil {
slog.ErrorContext(ctx, "Task Status Error Begin Transaction", "err", err) slog.ErrorContext(ctx, "Task Status Error Begin Transaction", "err", err2)
return return
} }
defer tx.Rollback(ctx) defer tx.Rollback(ctx)
@ -216,47 +216,47 @@ func updateWorkerTaskStatus(ctx context.Context) {
// 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", taskID) slog.ErrorContext(ctx, "Task is unknown by worker, Failing...", "err", err, "id", taskID)
_, err = tx.Exec(ctx, "UPDATE tasks SET status = $2 WHERE id = $1", taskID, constants.TASK_STATUS_FAILED) _, err2 = tx.Exec(ctx, "UPDATE tasks SET status = $2 WHERE id = $1", taskID, constants.TASK_STATUS_FAILED)
if err != nil { if err2 != nil {
slog.ErrorContext(ctx, "Updating Failed Task Status", "err", err) slog.ErrorContext(ctx, "Updating Failed Task Status", "err", err2)
return return
} }
_, err = tx.Exec(ctx, "INSERT INTO logs(task_id, message) VALUES($1,$2)", taskID, fmt.Sprintf("%v MASTER: Task Failed because it is unknown to Assigned Worker", time.Now())) _, err2 = tx.Exec(ctx, "INSERT INTO logs(task_id, message) VALUES($1,$2)", taskID, fmt.Sprintf("%v MASTER: Task Failed because it is unknown to Assigned Worker", time.Now()))
if err != nil { if err2 != nil {
slog.ErrorContext(ctx, "Updating Failed Task Log", "err", err) slog.ErrorContext(ctx, "Updating Failed Task Log", "err", err2)
return return
} }
slog.Info("Updating task done", "id", taskID, "status", constants.TASK_STATUS_FAILED) slog.Info("Updating task done", "id", taskID, "status", constants.TASK_STATUS_FAILED)
} 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
_, err = tx.Exec(ctx, "UPDATE tasks SET status = $2 WHERE id = $1", taskID, constants.TASK_STATUS_UNKNOWN) _, err2 = tx.Exec(ctx, "UPDATE tasks SET status = $2 WHERE id = $1", taskID, constants.TASK_STATUS_UNKNOWN)
if err != nil { if err2 != nil {
slog.ErrorContext(ctx, "Updating Unknown Task Status due to Timeout", "err", err) slog.ErrorContext(ctx, "Updating Unknown Task Status due to Timeout", "err", err2)
return return
} }
_, err = tx.Exec(ctx, "INSERT INTO logs(task_id, message) VALUES($1,$2)", taskID, fmt.Sprintf("%v MASTER: Status RPC Call Timed Out!", time.Now())) _, err2 = tx.Exec(ctx, "INSERT INTO logs(task_id, message) VALUES($1,$2)", taskID, fmt.Sprintf("%v MASTER: Status RPC Call Timed Out!", time.Now()))
if err != nil { if err2 != nil {
slog.ErrorContext(ctx, "Updating Unknown Task Log due to Timeout", "err", err) slog.ErrorContext(ctx, "Updating Unknown Task Log due to Timeout", "err", err2)
return return
} }
} else { } else {
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", taskID, constants.TASK_STATUS_UNKNOWN) _, err2 = db.Exec(ctx, "UPDATE tasks SET status = $2 WHERE id = $1", taskID, constants.TASK_STATUS_UNKNOWN)
if err != nil { if err2 != nil {
slog.ErrorContext(ctx, "Updating Unknown Task Status", "err", err) slog.ErrorContext(ctx, "Updating Unknown Task Status", "err", err2)
return return
} }
_, err = tx.Exec(ctx, "INSERT INTO logs(task_id, message) VALUES($1,$2)", taskID, fmt.Sprintf("%v MASTER: Status RPC Call Error: %v", time.Now(), err.Error())) _, err2 = tx.Exec(ctx, "INSERT INTO logs(task_id, message) VALUES($1,$2)", taskID, fmt.Sprintf("%v MASTER: Status RPC Call Error: %v", time.Now(), err.Error()))
if err != nil { if err2 != nil {
slog.ErrorContext(ctx, "Updating Unknown Task Log", "err", err) slog.ErrorContext(ctx, "Updating Unknown Task Log", "err", err2)
return return
} }
} }
err = tx.Commit(ctx) err2 = tx.Commit(ctx)
if err != nil { if err2 != nil {
slog.ErrorContext(ctx, "Task Status Error Commit Transaction", "err", err) slog.ErrorContext(ctx, "Task Status Error Commit Transaction", "err", err2)
} }
return return
} }