Refine queue worker pool behavior and surface worker configuration

.env.example

@@ -7,6 +7,7 @@ SMTP_HEALTH_ADDR=:8080
 SMTP_HEALTH_PORT=
 SMTP_HEALTH_DISABLE=false
 SMTP_QUEUE_PATH=./data/spool
+SMTP_QUEUE_WORKERS=
 
 # Access control
 SMTP_ALLOW_NETWORKS=127.0.0.1/32

CHANGELOG.md

@@ -29,6 +29,11 @@
 - Website: Add dedicated contact section with email and GitHub links.
 - Docs: Consolidate feature list in README (single "Features" section).
 - Website: Mirror unified feature list on the marketing page.
+- SMTP: Send a 554 5.7.1 rejection prior to closing unauthorized sessions so operators see explicit failures when hosts are misconfigured.
+- Storage: Roll back partially persisted messages when any recipient write fails to keep the spool aligned with the delivery queue.
+- Queue: Introduce a configurable worker pool with idempotent shutdown semantics and tests that verify real concurrency instead of timing heuristics.
+- Config: Document and surface the new `SMTP_QUEUE_WORKERS` environment variable across README, `.env.example`, install tooling, and the marketing site.
+- Brand: Refresh GopherPost logo and favicon with updated gopher-and-envelope concept; refine site header hover styling.
 - Brand: Refresh GopherPost logo and favicon with updated gopher-and-envelope concept; add site styling for ringed logo hover state.
 
 ## v0.4.0

README.md

@@ -9,6 +9,7 @@ Current release: `v0.4.0`
 - Enforces allow-listed access by host/IP before any banner is sent, adding a coarse ingress control layer for the unauthenticated listener.
 - Persists accepted messages to disk with per-recipient hashing so stored artefacts are private yet available for later inspection or reprocessing.
 - Performs outbound delivery via MX resolution, randomised equal-priority retries, opportunistic STARTTLS, and jittered exponential backoff managed by the in-memory queue.
+- Scales queue throughput with a configurable pool of concurrent delivery workers so busy deployments keep pace with inbound traffic.
 - Optionally signs outbound mail with DKIM when selector, key, and domain values are supplied through environment variables.
 - Provides built-in observability: a health server exposes readiness, `/metrics` instrumentation, and an optional live audit log stream; audit logging can be toggled at runtime and fanned out to subscribers.
 - Loads configuration entirely from environment variables (with `.env` support) covering ports, banner text, TLS/DKIM assets, and queue storage paths, simplifying containerised or systemd deployments.
@@ -45,9 +46,10 @@ SMTP_HOSTNAME # Hostname advertised in SMTP banners and HELO/EHLO (default syste
 SMTP_BANNER # Custom greeting appended to the initial 220 response (default GopherPost ready).  
 SMTP_DEBUG # Enable verbose audit logging when `true` (default `false`).  
 SMTP_HEALTH_ADDR # Listen address for the health server (default :8080).  
-SMTP_HEALTH_PORT # Override only the port component of the health address (e.g. 9090).  
-SMTP_HEALTH_DISABLE # Disable the health endpoint when `true` (default `false`).  
-SMTP_QUEUE_PATH # Directory used to persist inbound messages (default ./data/spool).  
+SMTP_HEALTH_PORT # Override only the port component of the health address (e.g. 9090).
+SMTP_HEALTH_DISABLE # Disable the health endpoint when `true` (default `false`).
+SMTP_QUEUE_PATH # Directory used to persist inbound messages (default ./data/spool).
+SMTP_QUEUE_WORKERS # Number of concurrent delivery workers processing the outbound queue (default logical CPU count).
 ```
 #### Access control
 

docs/index.html

@@ -118,6 +118,7 @@ <h2>Features</h2>
               <li>Enforces host/IP allow-lists before sending the banner for reliable ingress control.</li>
               <li>Persists accepted messages with per-recipient hashing so stored artefacts stay private yet reviewable.</li>
               <li>Delivers outbound mail via MX resolution, randomized equal-priority retries, opportunistic STARTTLS, and jittered exponential backoff.</li>
+              <li>Scales with configurable concurrent delivery workers so the queue keeps up with bursty inbound traffic.</li>
               <li>Supports DKIM signing when selector, key, and domain values are configured.</li>
               <li>Provides observability through the health server, `/metrics`, and optional live audit streaming with runtime-configurable logging.</li>
               <li>Configures entirely via environment variables (with `.env` support) for ports, banners, TLS/DKIM assets, and queue storage paths.</li>

install.sh

@@ -128,6 +128,7 @@ SMTP_HEALTH_PORT="$(ask SMTP_HEALTH_PORT 'SMTP_HEALTH_PORT' '8877')"
 SMTP_HEALTH_DISABLE="$(ask SMTP_HEALTH_DISABLE 'SMTP_HEALTH_DISABLE (true/false)' 'false')"
 
 SMTP_QUEUE_PATH="$(ask SMTP_QUEUE_PATH 'SMTP_QUEUE_PATH' "${DEFAULT_QUEUE_DIR}")"
+SMTP_QUEUE_WORKERS="$(ask SMTP_QUEUE_WORKERS 'SMTP_QUEUE_WORKERS (concurrent workers, blank for auto)' '')"
 SMTP_ALLOW_NETWORKS="$(ask SMTP_ALLOW_NETWORKS 'SMTP_ALLOW_NETWORKS (CIDR list)' '')"
 SMTP_ALLOW_HOSTS="$(ask SMTP_ALLOW_HOSTS 'SMTP_ALLOW_HOSTS (comma-separated)' '127.0.0.1')"
 SMTP_REQUIRE_LOCAL_DOMAIN="$(ask SMTP_REQUIRE_LOCAL_DOMAIN 'SMTP_REQUIRE_LOCAL_DOMAIN (true/false)' 'true')"
@@ -171,6 +172,7 @@ env_lines+=("Environment=\"SMTP_DEBUG=$(esc_env_val "$SMTP_DEBUG")\"")
 env_lines+=("Environment=\"SMTP_HEALTH_PORT=$(esc_env_val "$SMTP_HEALTH_PORT")\"")
 env_lines+=("Environment=\"SMTP_HEALTH_DISABLE=$(esc_env_val "$SMTP_HEALTH_DISABLE")\"")
 env_lines+=("Environment=\"SMTP_QUEUE_PATH=$(esc_env_val "$SMTP_QUEUE_PATH")\"")
+[[ -n "$SMTP_QUEUE_WORKERS" ]] && env_lines+=("Environment=\"SMTP_QUEUE_WORKERS=$(esc_env_val "$SMTP_QUEUE_WORKERS")\"")
 env_lines+=("Environment=\"SMTP_ALLOW_NETWORKS=$(esc_env_val "$SMTP_ALLOW_NETWORKS")\"")
 env_lines+=("Environment=\"SMTP_ALLOW_HOSTS=$(esc_env_val "$SMTP_ALLOW_HOSTS")\"")
 env_lines+=("Environment=\"SMTP_REQUIRE_LOCAL_DOMAIN=$(esc_env_val "$SMTP_REQUIRE_LOCAL_DOMAIN")\"")

internal/config/env_test.go

@@ -1,6 +1,9 @@
 package config
 
-import "testing"
+import (
+	"runtime"
+	"testing"
+)
 
 func TestBool(t *testing.T) {
 	t.Setenv("BOOL_TRUE", "true")
@@ -20,3 +23,26 @@ func TestBool(t *testing.T) {
 		t.Fatalf("unexpected override for unsupported values")
 	}
 }
+
+func TestQueueWorkers(t *testing.T) {
+	t.Setenv("SMTP_QUEUE_WORKERS", "")
+	expectedDefault := runtime.NumCPU()
+	if got := QueueWorkers(); got != expectedDefault {
+		t.Fatalf("expected default workers %d, got %d", expectedDefault, got)
+	}
+
+	t.Setenv("SMTP_QUEUE_WORKERS", "3")
+	if got := QueueWorkers(); got != 3 {
+		t.Fatalf("expected configured workers 3, got %d", got)
+	}
+
+	t.Setenv("SMTP_QUEUE_WORKERS", "-5")
+	if got := QueueWorkers(); got != expectedDefault {
+		t.Fatalf("expected fallback to default for negative value, got %d", got)
+	}
+
+	t.Setenv("SMTP_QUEUE_WORKERS", "noise")
+	if got := QueueWorkers(); got != expectedDefault {
+		t.Fatalf("expected fallback to default for invalid value, got %d", got)
+	}
+}

internal/config/queue.go

@@ -0,0 +1,22 @@
+package config
+
+import (
+	"os"
+	"runtime"
+	"strconv"
+	"strings"
+)
+
+// QueueWorkers returns the configured number of concurrent delivery workers.
+// Defaults to the number of logical CPUs when unset or invalid.
+func QueueWorkers() int {
+	value := strings.TrimSpace(os.Getenv("SMTP_QUEUE_WORKERS"))
+	if value == "" {
+		return runtime.NumCPU()
+	}
+	workers, err := strconv.Atoi(value)
+	if err != nil || workers < 1 {
+		return runtime.NumCPU()
+	}
+	return workers
+}

main.go

@@ -19,30 +19,30 @@ import (
 
 	"github.com/joho/godotenv"
 
-    health "gopherpost/health"
-    audit "gopherpost/internal/audit"
-    "gopherpost/internal/config"
-    "gopherpost/internal/dkim"
-    "gopherpost/internal/email"
-    "gopherpost/internal/metrics"
-    "gopherpost/internal/version"
-    "gopherpost/queue"
-    "gopherpost/storage"
-    tlsconfig "gopherpost/tlsconfig"
+	health "gopherpost/health"
+	audit "gopherpost/internal/audit"
+	"gopherpost/internal/config"
+	"gopherpost/internal/dkim"
+	"gopherpost/internal/email"
+	"gopherpost/internal/metrics"
+	"gopherpost/internal/version"
+	"gopherpost/queue"
+	"gopherpost/storage"
+	tlsconfig "gopherpost/tlsconfig"
 )
 
 const (
 	defaultSMTPPort   = "2525"
 	defaultHealthAddr = ":8080"
-    defaultBanner     = "GopherPost ready"
+	defaultBanner     = "GopherPost ready"
 	maxMessageBytes   = 10 << 20 // 10 MiB
 	commandDeadline   = 15 * time.Minute
 )
 
 func main() {
 	_ = godotenv.Load()
 	audit.RefreshFromEnv()
-    log.Printf("GopherPost version %s starting", version.Number)
+	log.Printf("GopherPost version %s starting", version.Number)
 	audit.Log("version %s boot", version.Number)
 
 	port := defaultSMTPPort
@@ -89,11 +89,14 @@ func main() {
 		log.Printf("Health endpoint listening on %s/healthz", healthListener.Addr().String())
 	}
 
-	q := queue.NewManager()
+	workerCount := config.QueueWorkers()
+	q := queue.NewManager(queue.WithWorkers(workerCount))
 	if dir := strings.TrimSpace(os.Getenv("SMTP_QUEUE_PATH")); dir != "" {
 		storage.SetBaseDir(dir)
 		log.Printf("Queue storage path set to %s", dir)
 	}
+	log.Printf("Queue workers configured: %d", workerCount)
+	audit.Log("queue workers %d", workerCount)
 	q.Start()
 	defer q.Stop()
 
@@ -141,7 +144,18 @@ func handleSession(conn net.Conn, q *queue.Manager, greeting string, hostname st
 		prefixArgs := append([]any{sessionID}, args...)
 		audit.Log("session %s "+format, prefixArgs...)
 	}
+	send := func(code int, msg string) bool {
+		t := fmt.Sprintf("%d %s", code, msg)
+		if err := tp.PrintfLine(t); err != nil {
+			log.Printf("send error to %s: %v", remote, err)
+			alog("send error: %v", err)
+			return false
+		}
+		alog("sent %d %s", code, msg)
+		return true
+	}
 	if !connAllowed(remoteAddr) {
+		_ = send(554, "5.7.1 Access denied")
 		audit.Log("session %s rejected remote %s", sessionID, remote)
 		return
 	}
@@ -159,17 +173,6 @@ func handleSession(conn net.Conn, q *queue.Manager, greeting string, hostname st
 		return
 	}
 
-	send := func(code int, msg string) bool {
-		t := fmt.Sprintf("%d %s", code, msg)
-		if err := tp.PrintfLine(t); err != nil {
-			log.Printf("send error to %s: %v", remote, err)
-			alog("send error: %v", err)
-			return false
-		}
-		alog("sent %d %s", code, msg)
-		return true
-	}
-
 	if !send(220, greeting) {
 		return
 	}
@@ -319,15 +322,18 @@ func handleSession(conn net.Conn, q *queue.Manager, greeting string, hostname st
 			}
 			payload := queue.NewPayload(messageBytes)
 			var queued []queue.QueuedMessage
+			var persistedPaths []string
 			var persistErr error
 
 			for _, rcpt := range to {
-				if err := storage.SaveMessage(messageID, from, rcpt, messageBytes); err != nil {
+				path, err := storage.SaveMessage(messageID, from, rcpt, messageBytes)
+				if err != nil {
 					log.Printf("failed to persist message for %s: %v", rcpt, err)
 					alog("storage error for %s: %v", rcpt, err)
 					persistErr = err
 					break
 				}
+				persistedPaths = append(persistedPaths, path)
 				queued = append(queued, queue.QueuedMessage{
 					ID:      messageID,
 					From:    from,
@@ -336,6 +342,12 @@ func handleSession(conn net.Conn, q *queue.Manager, greeting string, hostname st
 				})
 			}
 			if persistErr != nil {
+				for _, path := range persistedPaths {
+					if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
+						log.Printf("failed to roll back persisted message %s: %v", path, err)
+						alog("rollback error %s: %v", path, err)
+					}
+				}
 				if !send(451, "Requested action aborted: storage failure") {
 					return
 				}