Updated partition options

pgqueue/manager/manager.go

@@ -79,7 +79,7 @@ func New(ctx context.Context, pool pg.PoolConn, opts ...Opt) (*Manager, error) {
 
 	// Register a maintenance ticker
 	if _, err := self.RegisterTicker(bootstrapCtx, schema.DefaultMaintenanceTickerName, schema.TickerMeta{
-		Interval: types.Ptr(schema.DefaultMaintenancePeriod),
+		Interval: types.Ptr(self.maintenancePeriod),
 	}, self.maintenance); err != nil {
 		endBootstrapSpan(err)
 		return nil, err

pgqueue/manager/opt.go

@@ -2,6 +2,7 @@ package manager
 
 import (
 	"os"
+	"time"
 
 	// Packages
 	schema "github.com/mutablelogic/go-pg/pgqueue/schema"
@@ -17,10 +18,14 @@ type Opt func(*opt) error
 
 // opt combines all configuration options for Manager.
 type opt struct {
-	worker  string
-	schema  string
-	tracer  trace.Tracer
-	metrics metric.Meter
+	worker             string
+	schema             string
+	tracer             trace.Tracer
+	metrics            metric.Meter
+	partitionSize      uint64
+	partitionThreshold float64
+	partitionAhead     uint64
+	maintenancePeriod  time.Duration
 }
 
 ///////////////////////////////////////////////////////////////////////////////
@@ -40,6 +45,10 @@ func (o *opt) apply(opts ...Opt) error {
 
 func (o *opt) defaults() error {
 	o.schema = schema.DefaultSchema
+	o.partitionSize = schema.DefaultPartitionSize
+	o.partitionThreshold = schema.DefaultPartitionThreshold
+	o.partitionAhead = schema.DefaultPartitionAhead
+	o.maintenancePeriod = schema.DefaultMaintenancePeriod
 	if hostname, err := os.Hostname(); err != nil {
 		return err
 	} else {
@@ -86,3 +95,43 @@ func WithMeter(meter metric.Meter) Opt {
 		return nil
 	}
 }
+
+// WithPartitionSize sets the task partition size. Values less than 1 are ignored.
+func WithPartitionSize(size uint64) Opt {
+	return func(o *opt) error {
+		if size > 0 {
+			o.partitionSize = size
+		}
+		return nil
+	}
+}
+
+// WithPartitionThreshold sets the partition creation threshold in the range (0,1].
+func WithPartitionThreshold(threshold float64) Opt {
+	return func(o *opt) error {
+		if threshold > 0 && threshold <= 1 {
+			o.partitionThreshold = threshold
+		}
+		return nil
+	}
+}
+
+// WithPartitionAhead sets how many partitions to create when threshold is reached.
+func WithPartitionAhead(ahead uint64) Opt {
+	return func(o *opt) error {
+		if ahead > 0 {
+			o.partitionAhead = ahead
+		}
+		return nil
+	}
+}
+
+// WithMaintenancePeriod sets how often the maintenance ticker runs. Values less than 1 second are ignored.
+func WithMaintenancePeriod(period time.Duration) Opt {
+	return func(o *opt) error {
+		if period >= time.Second {
+			o.maintenancePeriod = period
+		}
+		return nil
+	}
+}

pgqueue/manager/partition.go

@@ -38,26 +38,41 @@ func (manager *Manager) CreateNextPartition(ctx context.Context) (result string,
 		}
 	}
 
-	// Create next partition if no partitions exist, or seq is within 80% of upper bound
-	if maxEnd == 0 || seq >= uint64(float64(maxEnd)*0.8) {
-		start := maxEnd
-		if start == 0 {
-			start = 1
+	// Create partition(s) if no partitions exist, or sequence has crossed threshold.
+	if maxEnd == 0 || seq >= uint64(float64(maxEnd)*manager.partitionThreshold) {
+		count := manager.partitionAhead
+		if count == 0 {
+			count = 1
 		}
-		end := start + schema.DefaultPartitionSize
-		name := fmt.Sprintf("task_%08d_%08d", start, end)
-		if err := manager.CreatePartition(ctx, schema.PartitionMeta{
-			Partition: name,
-			Start:     start,
-			End:       end,
-		}); err != nil {
-			return "", err
+
+		created := make([]string, 0, count)
+		for i := uint64(0); i < count; i++ {
+			start := maxEnd
+			if start == 0 {
+				start = 1
+			}
+			end := start + manager.partitionSize
+			name := fmt.Sprintf("task_%08d_%08d", start, end)
+			if err := manager.CreatePartition(ctx, schema.PartitionMeta{
+				Partition: name,
+				Start:     start,
+				End:       end,
+			}); err != nil {
+				return "", err
+			}
+
+			created = append(created, name)
+			maxEnd = end
 		}
 
-		// Reset the connection
+		// Reset the connection to refresh partition metadata.
 		manager.PoolConn.Reset()
 
-		return name, nil
+		if len(created) == 1 {
+			return created[0], nil
+		}
+
+		return fmt.Sprintf("%s (+%d more)", created[0], len(created)-1), nil
 	}
 
 	return "", nil

pgqueue/manager/partition_test.go

@@ -2,9 +2,12 @@ package manager_test
 
 import (
 	"encoding/json"
+	"fmt"
 	"testing"
+	"time"
 
 	// Packages
+	manager "github.com/mutablelogic/go-pg/pgqueue/manager"
 	schema "github.com/mutablelogic/go-pg/pgqueue/schema"
 	test "github.com/mutablelogic/go-pg/pgqueue/test"
 	assert "github.com/stretchr/testify/assert"
@@ -94,6 +97,44 @@ func TestGetPartitionSeq(t *testing.T) {
 	assert.GreaterOrEqual(t, partition.Count, uint64(2))
 }
 
+func TestCreateNextPartitionThresholdAndAhead(t *testing.T) {
+	shared, ctx := test.Begin(t)
+	defer test.End(t)
+
+	schemaName := fmt.Sprintf("pgq_partition_%d", time.Now().UnixNano())
+	mgr, err := manager.New(ctx, shared.PoolConn,
+		manager.WithSchema(schemaName),
+		manager.WithPartitionSize(10),
+		manager.WithPartitionThreshold(0.5),
+		manager.WithPartitionAhead(2),
+	)
+	require.NoError(t, err)
+
+	partitions, err := mgr.ListPartitions(ctx)
+	require.NoError(t, err)
+	require.Len(t, partitions, 2)
+
+	queue, err := mgr.RegisterQueue(ctx, "partition_threshold_queue", schema.QueueMeta{}, noopTask)
+	require.NoError(t, err)
+	defer func() {
+		_, _ = mgr.DeleteQueue(ctx, queue.Queue)
+	}()
+
+	for i := 0; i < 11; i++ {
+		var taskID schema.TaskId
+		err := mgr.With("id", queue.Queue).Insert(ctx, &taskID, schema.TaskMeta{Payload: json.RawMessage(`{"n":1}`)})
+		require.NoError(t, err)
+	}
+
+	created, err := mgr.CreateNextPartition(ctx)
+	require.NoError(t, err)
+	assert.NotEmpty(t, created)
+
+	partitions, err = mgr.ListPartitions(ctx)
+	require.NoError(t, err)
+	assert.Len(t, partitions, 4)
+}
+
 func findPartitionByName(partitions []schema.Partition, name string) (schema.Partition, bool) {
 	for _, partition := range partitions {
 		if partition.Partition == name {

pgqueue/schema/schema.go

@@ -20,10 +20,12 @@ const (
 	QueueListLimit               = 100
 	TickerListLimit              = 100
 	DefaultPartitionSize         = 100_000 // tasks per partition
+	DefaultPartitionThreshold    = 0.5     // create partition(s) when sequence reaches this fraction of the highest partition end
+	DefaultPartitionAhead        = 1       // number of partition(s) to create when threshold is reached
 	DefaultMaintenanceTickerName = "$maintenance$"
 	DefaultCleanupTickerName     = "$cleanup$"
 	DefaultTickerPeriod          = 5 * time.Second  // how often to look for matured tickers
 	DefaultQueuePeriod           = 10 * time.Second // how often to poll queues for retries and missed notifications
 	DefaultCleanupPeriod         = 15 * time.Minute // how often to delete expired tasks
-	DefaultMaintenancePeriod     = time.Hour        // create and drop partitions
+	DefaultMaintenancePeriod     = 10 * time.Minute // create and drop partitions
 )