Updated

Author: djthorpe

pkg/pgqueue/config/task.go

@@ -13,7 +13,7 @@ import (
 	pgqueue "github.com/mutablelogic/go-server/pkg/pgqueue"
 	schema "github.com/mutablelogic/go-server/pkg/pgqueue/schema"
 	ref "github.com/mutablelogic/go-server/pkg/ref"
-	"github.com/mutablelogic/go-server/pkg/types"
+	types "github.com/mutablelogic/go-server/pkg/types"
 )
 
 ////////////////////////////////////////////////////////////////////////////////

pkg/pgqueue/manager.go

@@ -216,7 +216,6 @@ func (manager *Manager) RunTickerLoop(ctx context.Context, namespace string, ch
 	delta := schema.TickerPeriod
 	timer := time.NewTimer(100 * time.Millisecond)
 	defer timer.Stop()
-
 	prev := time.Now()
 
 	// Loop until context is cancelled

pkg/pgqueue/schema/ticker.go

@@ -224,17 +224,17 @@ const (
 	`
 	tickerNextFunc = `
         -- Return the next matured ticker for a namespace
-        CREATE OR REPLACE FUNCTION ${"schema"}.ticker_next(ns TEXT) RETURNS TABLE (
+        CREATE OR REPLACE FUNCTION ${"schema"}.ticker_next(namespace TEXT) RETURNS TABLE (
             "ticker" TEXT, "interval" INTERVAL, "ns" TEXT, "ts" TIMESTAMP
         ) AS $$
 			WITH 
-				next_ticker AS (` + tickerSelect + `WHERE "ns" = ns AND ("ts" IS NULL OR "ts" + "interval" < NOW()) ORDER BY "ts" NULLS FIRST)
+				next_ticker AS (` + tickerSelect + `WHERE "ns" = namespace AND ("ts" IS NULL OR "ts" + "interval" <= NOW()) ORDER BY "ts" NULLS FIRST)
 			UPDATE
 				${"schema"}.ticker
 			SET
 				"ts" = TIMEZONE('UTC', NOW())
 			WHERE
-				"ns" = ns AND "ticker" = (SELECT "ticker" FROM next_ticker LIMIT 1 FOR UPDATE SKIP LOCKED)
+				"ns" = namespace AND "ticker" = (SELECT "ticker" FROM next_ticker LIMIT 1 FOR UPDATE SKIP LOCKED)
 			RETURNING		
 				"ticker", "interval", "ns", "ts"
         $$ LANGUAGE SQL