feat: specialized queue workers for prover server (#1778)

* Add specialized queue workers and Docker deployment support

- Replace generic queue workers with circuit-specific workers (update, append, address-append)
- Add Docker deployment files (Dockerfile.light, docker-compose.yml)
- Implement separate cleanup routines for old requests (30min) and results (1000+ items)
- Add comprehensive Redis queue test suite with 1000+ lines of tests
- Remove priority system and queue-workers flag in favor of circuit-based routing
- Always route batch operations to queues to prevent cross-contamination
- Enhanced failed job status reporting with detailed error information
- Fix SERVER_MODE environment variable typo

* Add Redis service and tests to prover workflow

* Remove Redis connectivity check from prover tests

Author: sergeytimoshin

.github/workflows/prover-test.yml

@@ -26,6 +26,16 @@ jobs:
     if: github.event.pull_request.draft == false
     runs-on: buildjet-8vcpu-ubuntu-2204
     timeout-minutes: 120
+    services:
+      redis:
+        image: redis:7-alpine
+        ports:
+          - 6379:6379
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
     steps:
       - name: Checkout sources
         uses: actions/checkout@v4
@@ -68,6 +78,44 @@ jobs:
           cd prover/server
           go test ./prover -timeout 60m
 
+      - name: Redis Queue tests
+        env:
+          TEST_REDIS_URL: redis://localhost:6379/15
+        run: |
+          cd prover/server
+          go test -v -run TestRedis -timeout 10m
+
+      - name: Queue cleanup tests
+        env:
+          TEST_REDIS_URL: redis://localhost:6379/15
+        run: |
+          cd prover/server
+          go test -v -run TestCleanup -timeout 5m
+
+      - name: Worker selection tests
+        run: |
+          cd prover/server
+          go test -v -run TestWorkerSelection -timeout 5m
+
+      - name: Batch operations queue routing tests
+        run: |
+          cd prover/server
+          go test -v -run TestBatchOperations -timeout 5m
+
+      - name: Queue processing flow tests
+        env:
+          TEST_REDIS_URL: redis://localhost:6379/15
+        run: |
+          cd prover/server
+          go test -v -run TestJobProcessingFlow -timeout 5m
+
+      - name: Failed job status tests
+        env:
+          TEST_REDIS_URL: redis://localhost:6379/15
+        run: |
+          cd prover/server
+          go test -v -run TestFailedJobStatus -timeout 5m
+
       - name: Lightweight integration tests
         if: ${{ github.event.pull_request.base.ref == 'main' }}
         run: |

prover/server/Dockerfile.light

@@ -0,0 +1,26 @@
+FROM golang:1.20.3-alpine AS builder
+
+WORKDIR /app
+
+COPY go.mod go.sum ./
+RUN go mod download && go mod verify
+
+COPY . .
+
+ENV CGO_ENABLED=0
+RUN go build -v -o /usr/local/bin/light-prover .
+
+RUN mkdir -p /tmp/empty_proving_keys
+
+FROM gcr.io/distroless/base-debian11:nonroot
+
+COPY --from=builder /usr/local/bin/light-prover /usr/local/bin/light-prover
+
+WORKDIR /proving-keys
+
+COPY --chown=nonroot:nonroot --from=builder /tmp/empty_proving_keys /proving-keys/
+
+WORKDIR /
+
+ENTRYPOINT [ "light-prover" ]
+CMD [ "start" ]

prover/server/docker-compose.yml

@@ -0,0 +1,22 @@
+services:
+  redis:
+    image: redis:7.4.4-alpine3.21
+    container_name: redis
+    ports:
+      - "6379:6379"
+    restart: unless-stopped
+
+  prover:
+    image: sergeytimoshin/prover-light:1.0.0
+    container_name: prover
+    ports:
+      - "3001:3001"
+    volumes:
+      - ./proving-keys:/proving-keys/:ro
+    command: >
+      start
+      --run-mode forester-test
+      --redis-url=redis://redis:6379
+    depends_on:
+      - redis
+    restart: unless-stopped

prover/server/main.go

@@ -444,11 +444,6 @@ func runCli() {
 						Usage: "Run only HTTP server (no queue workers)",
 						Value: false,
 					},
-					&cli.IntFlag{
-						Name:  "queue-workers",
-						Usage: "Number of queue worker goroutines",
-						Value: 1,
-					},
 				},
 				Action: func(context *cli.Context) error {
 					if context.Bool("json-logging") {
@@ -481,14 +476,13 @@ func runCli() {
 
 					queueOnly := context.Bool("queue-only")
 					serverOnly := context.Bool("server-only")
-					numWorkers := context.Int("queue-workers")
 
 					enableQueue := redisURL != "" && !serverOnly
 					enableServer := !queueOnly
 
 					if os.Getenv("QUEUE_MODE") == "true" {
 						enableQueue = true
-						if os.Getenv("SERVER _MODE") != "true" {
+						if os.Getenv("SERVER_MODE") != "true" {
 							enableServer = false
 						}
 					}
@@ -497,10 +491,9 @@ func runCli() {
 						Bool("enable_queue", enableQueue).
 						Bool("enable_server", enableServer).
 						Str("redis_url", redisURL).
-						Int("queue_workers", numWorkers).
 						Msg("Starting ZK Prover service")
 
-					var workers []*server.QueueWorker
+					var workers []server.QueueWorker
 					var redisQueue *server.RedisQueue
 					var instance server.RunningJob
 
@@ -514,22 +507,49 @@ func runCli() {
 							return fmt.Errorf("failed to connect to Redis: %w", err)
 						}
 
-						startResultCleanup(redisQueue)
+						startCleanupRoutines(redisQueue)
 
 						if stats, err := redisQueue.GetQueueStats(); err == nil {
 							logging.Logger().Info().Interface("initial_queue_stats", stats).Msg("Redis connection successful")
 						}
 
-						if numWorkers <= 0 {
-							numWorkers = 1
+						logging.Logger().Info().Msg("Starting queue workers")
+
+						startAllWorkers := runMode == prover.Forester || runMode == prover.ForesterTest
+
+						var workersStarted []string
+
+						// Start update worker for batch-update circuits or forester modes
+						if startAllWorkers || containsCircuit(circuits, "update") || containsCircuit(circuits, "update-test") {
+							updateWorker := server.NewUpdateQueueWorker(redisQueue, psv1, psv2)
+							workers = append(workers, updateWorker)
+							go updateWorker.Start()
+							workersStarted = append(workersStarted, "update")
 						}
 
-						logging.Logger().Info().Int("workers", numWorkers).Msg("Starting queue workers")
+						// Start append worker for batch-append circuits or forester modes
+						if startAllWorkers || containsCircuit(circuits, "append-with-proofs") || containsCircuit(circuits, "append-with-proofs-test") {
+							appendWorker := server.NewAppendQueueWorker(redisQueue, psv1, psv2)
+							workers = append(workers, appendWorker)
+							go appendWorker.Start()
+							workersStarted = append(workersStarted, "append")
+						}
 
-						for i := 0; i < numWorkers; i++ {
-							worker := server.NewQueueWorker(i+1, redisQueue, psv1, psv2)
-							workers = append(workers, worker)
-							go worker.Start()
+						// Start address append worker for address-append circuits or forester modes
+						if startAllWorkers || containsCircuit(circuits, "address-append") || containsCircuit(circuits, "address-append-test") {
+							addressAppendWorker := server.NewAddressAppendQueueWorker(redisQueue, psv1, psv2)
+							workers = append(workers, addressAppendWorker)
+							go addressAppendWorker.Start()
+							workersStarted = append(workersStarted, "address-append")
+						}
+
+						if len(workersStarted) == 0 {
+							logging.Logger().Warn().Msg("No queue workers started - no matching circuits found")
+						} else {
+							logging.Logger().Info().
+								Strs("workers_started", workersStarted).
+								Bool("forester_mode", startAllWorkers).
+								Msg("Queue workers started")
 						}
 					}
 
@@ -931,17 +951,68 @@ func debugProvingSystemKeys(keysDirPath string, runMode prover.RunMode, circuits
 	}
 }
 
-func startResultCleanup(redisQueue *server.RedisQueue) {
+func startCleanupRoutines(redisQueue *server.RedisQueue) {
+	logging.Logger().Info().Msg("Running immediate cleanup on startup")
+
+	if err := redisQueue.CleanupOldRequests(); err != nil {
+		logging.Logger().Error().
+			Err(err).
+			Msg("Failed to cleanup old proof requests on startup")
+	} else {
+		logging.Logger().Info().Msg("Startup cleanup of old proof requests completed")
+	}
+
+	if err := redisQueue.CleanupOldResults(); err != nil {
+		logging.Logger().Error().
+			Err(err).
+			Msg("Failed to cleanup old results on startup")
+	} else {
+		logging.Logger().Info().Msg("Startup cleanup of old results completed")
+	}
+
+	// Start cleanup for old proof requests (every 10 minutes)
+	go func() {
+		requestTicker := time.NewTicker(10 * time.Minute)
+		defer requestTicker.Stop()
+
+		logging.Logger().Info().Msg("Started old proof requests cleanup routine (every 10 minutes)")
+
+		for range requestTicker.C {
+			if err := redisQueue.CleanupOldRequests(); err != nil {
+				logging.Logger().Error().
+					Err(err).
+					Msg("Failed to cleanup old proof requests")
+			} else {
+				logging.Logger().Debug().Msg("Old proof requests cleanup completed")
+			}
+		}
+	}()
+
+	// Start less frequent cleanup for old results (every 1 hour)
 	go func() {
-		ticker := time.NewTicker(1 * time.Hour)
-		defer ticker.Stop()
+		resultTicker := time.NewTicker(1 * time.Hour)
+		defer resultTicker.Stop()
+
+		logging.Logger().Info().Msg("Started old results cleanup routine (every 1 hour)")
 
-		for range ticker.C {
+		for range resultTicker.C {
 			if err := redisQueue.CleanupOldResults(); err != nil {
 				logging.Logger().Error().
 					Err(err).
 					Msg("Failed to cleanup old results")
+			} else {
+				logging.Logger().Debug().Msg("Old results cleanup completed")
 			}
 		}
 	}()
 }
+
+// containsCircuit checks if the circuits slice contains the specified circuit
+func containsCircuit(circuits []string, circuit string) bool {
+	for _, c := range circuits {
+		if c == circuit {
+			return true
+		}
+	}
+	return false
+}