From 8ec2d13a74028187ac5e7eb4f8c8d50a3d31eca8 Mon Sep 17 00:00:00 2001 From: David Thorpe Date: Thu, 18 Jun 2026 10:02:05 +0200 Subject: [PATCH 1/2] Updated partition options --- pgqueue/manager/manager.go | 2 +- pgqueue/manager/opt.go | 57 ++++++++++++++++++++++++++++--- pgqueue/manager/partition.go | 45 ++++++++++++++++-------- pgqueue/manager/partition_test.go | 41 ++++++++++++++++++++++ pgqueue/schema/schema.go | 4 ++- 5 files changed, 128 insertions(+), 21 deletions(-) diff --git a/pgqueue/manager/manager.go b/pgqueue/manager/manager.go index 60dfc89..78604c7 100644 --- a/pgqueue/manager/manager.go +++ b/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 diff --git a/pgqueue/manager/opt.go b/pgqueue/manager/opt.go index 24b1755..0d4327f 100644 --- a/pgqueue/manager/opt.go +++ b/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 + } +} diff --git a/pgqueue/manager/partition.go b/pgqueue/manager/partition.go index 3c84f6f..cd01c8b 100644 --- a/pgqueue/manager/partition.go +++ b/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 diff --git a/pgqueue/manager/partition_test.go b/pgqueue/manager/partition_test.go index ed64eb0..5db7791 100644 --- a/pgqueue/manager/partition_test.go +++ b/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 { diff --git a/pgqueue/schema/schema.go b/pgqueue/schema/schema.go index 29791ac..b9ea22f 100644 --- a/pgqueue/schema/schema.go +++ b/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 ) From 0a8ffbcdcf60c9ca56e4aaa86c48309a49442b5f Mon Sep 17 00:00:00 2001 From: David Thorpe Date: Thu, 18 Jun 2026 10:15:11 +0200 Subject: [PATCH 2/2] Updated partition test --- pgqueue/manager/partition_test.go | 41 ------------------------------- 1 file changed, 41 deletions(-) diff --git a/pgqueue/manager/partition_test.go b/pgqueue/manager/partition_test.go index 5db7791..ed64eb0 100644 --- a/pgqueue/manager/partition_test.go +++ b/pgqueue/manager/partition_test.go @@ -2,12 +2,9 @@ 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" @@ -97,44 +94,6 @@ 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 {