7 files changed, 69 insertions(+), 40 deletions(-)

M email/worker.go
M go.mod
M go.sum
M server/server.go
M webhooks/legacy.go
M webhooks/legacy_test.go
M webhooks/queue.go

@@ 9,6 9,7 @@ import (
 	"log"
 	"net/http"
 	"os"
+	"strconv"
 	"strings"
 	"time"
 


@@ 228,8 229,16 @@ func NewQueue(conf ini.File) *Queue {
 		}
 	}
 
+	queueSize := 512
+	if s, ok := conf.Get("mail", "egress-queue-size"); ok {
+		var err error
+		if queueSize, err = strconv.Atoi(s); err != nil {
+			panic(fmt.Errorf("[mail]egress-queue-size: %w", err))
+		}
+	}
+
 	return &Queue{
-		Queue:        work.NewQueue("email"),
+		Queue:        work.NewQueue("email", queueSize),
 		smtpFrom:     addr,
 		ownerAddress: ownerAddr,
 		entity:       entity,

@@ 3,7 3,7 @@ module git.sr.ht/~sircmpwn/core-go
 go 1.22
 
 require (
-	git.sr.ht/~sircmpwn/dowork v0.0.0-20221010085743-46c4299d76a1
+	git.sr.ht/~sircmpwn/dowork v0.0.0-20241209140539-95719cfc0118
 	git.sr.ht/~sircmpwn/getopt v1.0.0
 	git.sr.ht/~sircmpwn/go-bare v0.0.0-20210406120253-ab86bc2846d9
 	github.com/99designs/gqlgen v0.17.36

@@ 1,5 1,5 @@
-git.sr.ht/~sircmpwn/dowork v0.0.0-20221010085743-46c4299d76a1 h1:EvPKkneKkF/f7zEgKPqIZVyj3jWO8zSmsBOvMhAGqMA=
-git.sr.ht/~sircmpwn/dowork v0.0.0-20221010085743-46c4299d76a1/go.mod h1:8neHEO3503w/rNtttnR0JFpQgM/GFhaafVwvkPsFIDw=
+git.sr.ht/~sircmpwn/dowork v0.0.0-20241209140539-95719cfc0118 h1:aKZ3Es4Tj6IKLGilwcPlLzeXwrz+/ZalVimyiX6bmdM=
+git.sr.ht/~sircmpwn/dowork v0.0.0-20241209140539-95719cfc0118/go.mod h1:8neHEO3503w/rNtttnR0JFpQgM/GFhaafVwvkPsFIDw=
 git.sr.ht/~sircmpwn/getopt v0.0.0-20191230200459-23622cc906b3/go.mod h1:wMEGFFFNuPos7vHmWXfszqImLppbc0wEhh6JBfJIUgw=
 git.sr.ht/~sircmpwn/getopt v1.0.0 h1:/pRHjO6/OCbBF4puqD98n6xtPEgE//oq5U8NXjP7ROc=
 git.sr.ht/~sircmpwn/getopt v1.0.0/go.mod h1:wMEGFFFNuPos7vHmWXfszqImLppbc0wEhh6JBfJIUgw=

@@ 260,16 260,24 @@ func (server *Server) WithMiddleware(
 
 // Add dowork task queues for this server to manage
 func (server *Server) WithQueues(queues ...*work.Queue) *Server {
-	ctx := context.Background()
-	ctx = config.Context(ctx, server.conf, server.service)
-	ctx = database.Context(ctx, server.db)
-	ctx = redis.Context(ctx, server.redis)
-	ctx = email.Context(ctx, server.email)
-	ctx = context.WithValue(ctx, serverCtxKey, server)
-
+	queueWorkers := 1
+	if n, ok := server.conf.Get(server.service, "queue-workers"); ok {
+		var err error
+		if queueWorkers, err = strconv.Atoi(n); err != nil {
+			panic(fmt.Errorf("[%s]queue-workers: %w", server.service, err))
+		}
+	}
 	server.queues = append(server.queues, queues...)
 	for _, queue := range queues {
-		queue.Start(ctx)
+		// Use a different context per worker to allow "goroutine-local"
+		// variables.
+		ctx := context.Background()
+		ctx = config.Context(ctx, server.conf, server.service)
+		ctx = database.Context(ctx, server.db)
+		ctx = redis.Context(ctx, server.redis)
+		ctx = email.Context(ctx, server.email)
+		ctx = context.WithValue(ctx, serverCtxKey, server)
+		queue.Start(ctx, queueWorkers)
 	}
 	return server
 }

@@ 9,12 9,14 @@ import (
 	"io/ioutil"
 	"log"
 	"net/http"
+	"strconv"
 	"strings"
 	"time"
 
 	"git.sr.ht/~sircmpwn/dowork"
 	sq "github.com/Masterminds/squirrel"
 	"github.com/google/uuid"
+	"github.com/vaughan0/go-ini"
 
 	"git.sr.ht/~sircmpwn/core-go/crypto"
 	"git.sr.ht/~sircmpwn/core-go/database"


@@ 33,9 35,16 @@ type LegacySubscription struct {
 
 // Creates a new worker for delivering legacy webhooks. The caller must start
 // the worker themselves.
-func NewLegacyQueue() *LegacyQueue {
+func NewLegacyQueue(conf ini.File) *LegacyQueue {
+	queueSize := 512
+	if s, ok := conf.Get("webhooks", "queue-size"); ok {
+		var err error
+		if queueSize, err = strconv.Atoi(s); err != nil {
+			panic(fmt.Errorf("[webhooks]queue-size: %w", err))
+		}
+	}
 	return &LegacyQueue{
-		work.NewQueue("webhooks_legacy"),
+		work.NewQueue("webhooks_legacy", queueSize),
 	}
 }
 

@@ 57,12 57,11 @@ func (ac *argContains) Match(v driver.Value) bool {
 }
 
 func TestDelivery(t *testing.T) {
-	var called bool
+	called := make(chan struct{})
 	srv := httptest.NewServer(http.HandlerFunc(
 		func(w http.ResponseWriter, r *http.Request) {
 			defer r.Body.Close()
 
-			called = true
 			assert.Equal(t, r.Method, http.MethodPost)
 			assert.Equal(t, r.URL.Path, "/webhook")
 


@@ 79,6 78,7 @@ func TestDelivery(t *testing.T) {
 			assert.True(t, crypto.VerifyWebhook(b, nonce, signature))
 
 			w.Write([]byte("Thanks!"))
+			close(called)
 		}))
 	defer srv.Close()
 


@@ 87,6 87,8 @@ func TestDelivery(t *testing.T) {
 		panic(err)
 	}
 	ctx := database.Context(context.Background(), db)
+	ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
+	defer cancel()
 
 	// Lookup phase
 	mock.ExpectBegin()


@@ 97,30 99,12 @@ func TestDelivery(t *testing.T) {
 						srv.URL+"/webhook", "profile:update")).
 		WithArgs(42, sqlmock.AnyArg()) // Any => events LIKE %profile:update%
 	mock.ExpectCommit()
-
-	queue := NewLegacyQueue()
-	q := sq.
-		Select().
-		From("user_webhook_subscription sub").
-		Where(`sub.user_id = ?`, 42)
-	queue.Schedule(ctx, q, "user", "profile:update", []byte(`{"hello": "world"}`))
-
 	// Schedule phase
 	mock.ExpectBegin()
 	mock.ExpectQuery(`INSERT INTO user_webhook_delivery`).
 		WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow(4096))
 	mock.ExpectCommit()
-
-	queue.Queue.Dispatch(ctx)
-
-	assert.Nil(t, mock.ExpectationsWereMet())
-
 	// Delivery phase
-	db, mock, err = sqlmock.New()
-	if err != nil {
-		panic(err)
-	}
-
 	mock.ExpectBegin()
 	mock.ExpectExec(`UPDATE user_webhook_delivery`).
 		WithArgs("Thanks!", 200,


@@ 135,9 119,19 @@ func TestDelivery(t *testing.T) {
 		WillReturnResult(sqlmock.NewResult(1, 1))
 	mock.ExpectCommit()
 
-	ctx = database.Context(context.Background(), db)
-	queue.Queue.Dispatch(ctx)
+	queue := NewLegacyQueue(make(ini.File))
+	queue.Queue.Start(ctx, 1)
+	q := sq.
+		Select().
+		From("user_webhook_subscription sub").
+		Where(`sub.user_id = ?`, 42)
+	queue.Schedule(ctx, q, "user", "profile:update", []byte(`{"hello": "world"}`))
 
-	assert.Nil(t, mock.ExpectationsWereMet())
-	assert.True(t, called)
+	select {
+	case <-called:
+		queue.Queue.Shutdown()
+		assert.Nil(t, mock.ExpectationsWereMet())
+	case <-ctx.Done():
+		t.Fatal("webhook url endpoint not called")
+	}
 }

@@ 9,6 9,7 @@ import (
 	"io/ioutil"
 	"log"
 	"net/http"
+	"strconv"
 	"strings"
 	"time"
 


@@ 16,6 17,7 @@ import (
 	"github.com/99designs/gqlgen/graphql"
 	sq "github.com/Masterminds/squirrel"
 	"github.com/google/uuid"
+	"github.com/vaughan0/go-ini"
 
 	"git.sr.ht/~sircmpwn/core-go/auth"
 	"git.sr.ht/~sircmpwn/core-go/crypto"


@@ 42,8 44,15 @@ type WebhookSubscription struct {
 
 // Creates a new worker for delivering webhooks. The caller must start the
 // worker themselves.
-func NewQueue(schema graphql.ExecutableSchema) *WebhookQueue {
-	return &WebhookQueue{work.NewQueue("webhooks"), schema}
+func NewQueue(schema graphql.ExecutableSchema, conf ini.File) *WebhookQueue {
+	queueSize := 512
+	if s, ok := conf.Get("webhooks", "queue-size"); ok {
+		var err error
+		if queueSize, err = strconv.Atoi(s); err != nil {
+			panic(fmt.Errorf("[webhooks]queue-size: %w", err))
+		}
+	}
+	return &WebhookQueue{work.NewQueue("webhooks", queueSize), schema}
 }
 
 // Schedules delivery of a webhook to a set of subscribers.