wip

migrations/000016_alter_workers_table_queue_enable.down.sql

@@ -1,2 +0,0 @@
-ALTER TABLE workers
-DROP IF EXISTS  queue_enable;

migrations/000016_alter_workers_table_queue_enable.up.sql

@@ -1,2 +0,0 @@
-ALTER TABLE workers
-ADD queue_enable boolean NOT NULL DEFAULT false;

server/index.go

@@ -16,10 +16,8 @@ type IndexData struct {
 }
 
 type IndexWorker struct {
-	ID string
 	Worker
 	Status *types.WorkerStatus
-	QueueEnable bool
 }
 
 func handleIndex(w http.ResponseWriter, r *http.Request) {
@@ -41,13 +39,11 @@ func handleIndex(w http.ResponseWriter, r *http.Request) {
 				}
 				slog.InfoContext(r.Context(), "Got Worker Status", "id", i, "status", status)
 				data.Workers = append(data.Workers, IndexWorker{
-					ID:     i,
 					Worker: *Workers[i],
 					Status: &status,
 				})
 			} else {
 				data.Workers = append(data.Workers, IndexWorker{
-					ID:     i,
 					Worker: *Workers[i],
 					Status: nil,
 				})

server/queue.go

@@ -1,40 +0,0 @@
-package server
-
-import (
-	"fmt"
-	"log/slog"
-	"net/http"
-)
-
-func HandleSetQueueEnable(w http.ResponseWriter, r *http.Request) {
-	err := r.ParseForm()
-	if err != nil {
-		http.Error(w, fmt.Sprintf("Parseing Form: %v", err), http.StatusBadRequest)
-		return
-	}
-	enable := r.FormValue("enable")
-	if enable != "true" && enable != "false" {
-		http.Error(w, "Enable must be true or false", http.StatusBadRequest)
-		return
-	}
-
-	worker := r.FormValue("worker")
-
-	slog.Info("Got set Queue Enable", "enable", enable, "worker", worker)
-
-	if worker == "all" {
-		_, err = db.Exec(r.Context(), "UPDATE workers SET queue_enable = $1", enable)
-		if err != nil {
-			http.Error(w, fmt.Sprintf("Setting Worker Queue Enable: %v", err), http.StatusInternalServerError)
-			return
-		}
-	} else {
-		_, err = db.Exec(r.Context(), "UPDATE workers SET queue_enable = $1 where id = $2", enable, worker)
-		if err != nil {
-			http.Error(w, fmt.Sprintf("Setting Worker Queue Enable: %v", err), http.StatusInternalServerError)
-			return
-		}
-	}
-
-	http.Redirect(w, r, r.Header.Get("Referer"), http.StatusFound)
-}

server/server.go

@@ -101,7 +101,6 @@ func Start(_conf config.Config, tmplFS embed.FS, staticFS embed.FS, migrationsFS
 	mux.HandleFunc("/libraries/{id}", handleLibrary)
 	mux.HandleFunc("/libraries", handleLibraries)
 	mux.HandleFunc("/ffmpeg_commands", handleFfmpegCommands)
-	mux.HandleFunc("/queue_enable", HandleSetQueueEnable)
 
 	mux.HandleFunc("/", handleIndex)
 

server/task.go

@@ -332,18 +332,7 @@ type QueuedTask struct {
 	FfmpegCommand types.FFmpegCommand `json:"ffmpeg_command" db:"ffmpeg_command"`
 }
 
-var assignRunning = false
-
 func assignQueuedTasks(ctx context.Context) error {
-	if assignRunning {
-		return nil
-	}
-
-	assignRunning = true
-	defer func() {
-		assignRunning = false
-	}()
-
 	rows, err := db.Query(ctx, "SELECT t.id as id, t.type as type, t.file_id as file_id, f.md5 as md5, t.data as data, fc.data as ffmpeg_command FROM tasks t INNER JOIN files f ON f.id = t.file_id INNER JOIN ffmpeg_commands fc ON fc.id = t.ffmpeg_command_id WHERE t.status = $1", constants.TASK_STATUS_QUEUED)
 	if err != nil {
 		return fmt.Errorf("Query Queued Tasks: %w", err)
@@ -371,23 +360,13 @@ func assignQueuedTasks(ctx context.Context) error {
 			return nil
 		}
 		if Workers[i].Connected {
-			var queueEnable bool
-			err := db.QueryRow(ctx, "SELECT queue_enable FROM workers WHERE id = $1", i).Scan(&queueEnable)
-			if err != nil {
-				return fmt.Errorf("Error Querying Worker Queue Enable: %w", err)
-			}
-			if !queueEnable {
-				slog.DebugContext(ctx, "Skipping Worker since Queueing is disabled", "worker_id", i)
-				continue
-			}
-
 			var count int
-			err = db.QueryRow(ctx, "SELECT COUNT(*) FROM tasks WHERE worker_id = $1 AND (status = $2 OR status = $3 OR status = $4 OR status = $5)", i, constants.TASK_STATUS_UNKNOWN, constants.TASK_STATUS_ASSIGNED, constants.TASK_STATUS_RUNNING, constants.TASK_STATUS_WAITING).Scan(&count)
+			err := db.QueryRow(ctx, "SELECT COUNT(*) FROM tasks WHERE worker_id = $1 AND (status = $2 OR status = $3 OR status = $4 OR status = $5)", i, constants.TASK_STATUS_UNKNOWN, constants.TASK_STATUS_ASSIGNED, constants.TASK_STATUS_RUNNING, constants.TASK_STATUS_WAITING).Scan(&count)
 			if err != nil {
 				return fmt.Errorf("Error Querying Worker Task Count: %w", err)
 			}
 
-			slog.Debug("Assigning Queued Tasks Worker", "worker", i, "count", count)
+			slog.Info("Assigning Queued Tasks Worker", "worker", i, "count", count)
 
 			// Allow for Multiple Tasks at once in the future
 			if count < 1 {
@@ -416,7 +395,7 @@ func assignQueuedTasks(ctx context.Context) error {
 						// 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 = tx.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 {
 							return fmt.Errorf("Updating Task status during already running error: %w", err)
 						}
@@ -424,14 +403,14 @@ func assignQueuedTasks(ctx context.Context) error {
 						// 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)
 
-						_, err = tx.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 {
 							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 = tx.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())})
+						_, 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)
 						}

server/worker.go

@@ -129,7 +129,7 @@ func readMessage(ctx context.Context, c *websocket.Conn) error {
 	if err != nil {
 		return err
 	}
-	slog.DebugContext(ctx, "Got Websocket Message", "type", t.String(), "data", data)
+	slog.InfoContext(ctx, "Got Websocket Message", "type", t.String(), "data", data)
 	rpcServer.HandleMessage(ctx, c, data)
 	return nil
 }

tmpl/index.tmpl

@@ -15,22 +15,6 @@
         {{end}}
     {{end}}
 </div>
-<h2>Set Queue Enable</h2>
-<form method="POST" action= "/queue_enable">
-    <label for="worker">Worker:</label>
-    <select id="worker" name="worker">
-        <option value="all">All</option>
-        {{range $w := .Workers}}
-        <option value="{{$w.ID}}">{{$w.Name}}</option>
-        {{end}}
-    </select>
-    <label for="enable">Enable</label>
-    <select id="enable" name="enable">
-        <option value="true">True</option>
-        <option value="false">False</option>
-    </select>
-    <input type="submit" value="Submit">
-</form>
 <h2>Workers</h2>
 <div class="workers">
     {{range $w := .Workers}}

worker/worker.go

@@ -116,7 +116,7 @@ func readMessage(ctx context.Context, c *websocket.Conn) error {
 	if err != nil {
 		return err
 	}
-	slog.DebugContext(ctx, "Got Websocket Message", "type", t.String(), "data", data)
+	slog.InfoContext(ctx, "Got Websocket Message", "type", t.String(), "data", data)
 	rpcServer.HandleMessage(ctx, c, data)
 	return nil
 }