fix scheduler

* switched to using a scheduler for each occasion instead of one global scheduler for everything, this helps futur scheduled jobs not colliding with scheduled requests (reqscheduler)
* fixed scheduler fetching > limit
* fixed incoming scheduled requests only considering queue when determining wether to add to heap  (queue) => incoming requests now check if earlier than latest heap request AND if they are the earliest request in the db

Author: CodeShellDev

go.mod

@@ -11,7 +11,7 @@ require (
 	github.com/codeshelldev/gotl/pkg/pretty v0.0.10
 	github.com/codeshelldev/gotl/pkg/query v0.0.4
 	github.com/codeshelldev/gotl/pkg/request v0.0.8
-	github.com/codeshelldev/gotl/pkg/scheduler v0.0.7
+	github.com/codeshelldev/gotl/pkg/scheduler v0.0.9
 	github.com/codeshelldev/gotl/pkg/server/http v0.0.3
 	github.com/codeshelldev/gotl/pkg/stringutils v0.0.8
 	github.com/codeshelldev/gotl/pkg/templating v0.0.4

go.sum

@@ -16,8 +16,8 @@ github.com/codeshelldev/gotl/pkg/query v0.0.4 h1:o2Oagx/s1wfNMqkh6GfR6wpsIVOFSDP
 github.com/codeshelldev/gotl/pkg/query v0.0.4/go.mod h1:Bg3tFzFq9xButTw0BSfGQhSmfAnFDrJamOcnX6Io4m4=
 github.com/codeshelldev/gotl/pkg/request v0.0.8 h1:sVVt2ADOTgZrna7RsqThwMKxYCuxlBE80s7kV90rARg=
 github.com/codeshelldev/gotl/pkg/request v0.0.8/go.mod h1:ngE6/OksRIclheFGfqJ6/2lBpzCm9sPe4p5JfGIg5kg=
-github.com/codeshelldev/gotl/pkg/scheduler v0.0.7 h1:6D16m1/DndhkIvoYMc26ebc9SySy1UQMc7W4QifdvvM=
-github.com/codeshelldev/gotl/pkg/scheduler v0.0.7/go.mod h1:sXEpRxbDc/JAN8WDxxq5+UxJf2dOQpKJIZyvORjIJGM=
+github.com/codeshelldev/gotl/pkg/scheduler v0.0.9 h1:8IFSPmyQehogwwo9vo3aAo3NxVY5aR8Hzrb8E+hMrDU=
+github.com/codeshelldev/gotl/pkg/scheduler v0.0.9/go.mod h1:sXEpRxbDc/JAN8WDxxq5+UxJf2dOQpKJIZyvORjIJGM=
 github.com/codeshelldev/gotl/pkg/server/http v0.0.3 h1:3232uPB2CljzUJadyrME7p0DaOCGz+vPVfPjnS788SE=
 github.com/codeshelldev/gotl/pkg/server/http v0.0.3/go.mod h1:/asx7ViJtwlBvLgObjI/tejm6lNDN1/B+/6BPImqDfc=
 github.com/codeshelldev/gotl/pkg/stringutils v0.0.8 h1:VKIuEYLJARDmHyhAbcMy1TsdxPdzsKlbQvgr1G4QE7s=

internals/db/reqdb.go

@@ -214,16 +214,31 @@ func (s *RequestSchedulerDB) CleanupDones(threshold time.Duration) error {
 func (s *RequestSchedulerDB) FetchNext(amount int, within time.Duration) ([]*ScheduledRequest, error) {
 	minRunAt := time.Now().Add(within).Unix()
 
-	rows, err := s.db.Query(`
-		SELECT id, method, url, created_at, run_at, request_headers, request_body
-		FROM scheduled_requests
-        WHERE status = ? AND run_at <= ?
-		ORDER BY run_at ASC
-		LIMIT ?`,
-		STATUS_PENDING,
-		minRunAt,
-		amount,
-	)
+	var err error
+	var rows *sql.Rows
+
+	if within == 0 {
+		rows, err = s.db.Query(`
+			SELECT id, method, url, created_at, run_at, request_headers, request_body
+			FROM scheduled_requests
+			WHERE status = ?
+			ORDER BY run_at ASC
+			LIMIT ?`,
+			STATUS_PENDING,
+			amount,
+		)
+	} else {
+		rows, err = s.db.Query(`
+			SELECT id, method, url, created_at, run_at, request_headers, request_body
+			FROM scheduled_requests
+			WHERE status = ? AND run_at <= ?
+			ORDER BY run_at ASC
+			LIMIT ?`,
+			STATUS_PENDING,
+			minRunAt,
+			amount,
+		)
+	}
 
 	if err != nil {
 		return nil, err

internals/scheduler/reqscheduler.go

@@ -2,6 +2,7 @@ package scheduler
 
 import (
 	"bytes"
+	"context"
 	"errors"
 	"io"
 	"net/http"
@@ -11,20 +12,33 @@ import (
 	"github.com/codeshelldev/gotl/pkg/jsonutils"
 	"github.com/codeshelldev/gotl/pkg/logger"
 	"github.com/codeshelldev/gotl/pkg/request"
+	scheduling "github.com/codeshelldev/gotl/pkg/scheduler"
 	"github.com/codeshelldev/secured-signal-api/internals/db"
 	"github.com/google/uuid"
 )
 
 var rsdb *db.RequestSchedulerDB
 
+var reqscheduler = scheduling.New()
+var cancel context.CancelFunc
+
 const limit = 5
 const withinTime = 5 * time.Minute
 
 const recoveryThreshold = 10 * time.Minute
 
 const doneStaleThreshold = 24 * time.Hour
 
+func Stop() {
+	cancel()
+}
+
 func StartRequestScheduler() {
+	var ctx context.Context
+	ctx, cancel = context.WithCancel(context.Background())
+
+	go reqscheduler.Run(ctx)
+
 	rsdb = db.NewRequestSchedulerDB()
 
 	rsdb.CleanupDones(doneStaleThreshold)
@@ -36,15 +50,15 @@ func StartRequestScheduler() {
 		for range ticker.C {
 			rsdb.RecoverStales(recoveryThreshold)
 
-			if scheduler.Len() < limit {
+			if reqscheduler.Len() < limit {
 				UpdateQueue()
 			}
 		}
 	}()
 }
 
 func UpdateQueue() {
-	requests, _ := rsdb.FetchNext(limit, withinTime)
+	requests, _ := rsdb.FetchNext(limit - reqscheduler.Len(), withinTime)
 
 	for _, req := range requests {
 		AddToQueue(req)
@@ -55,24 +69,36 @@ func AddToQueue(req *db.ScheduledRequest) {
 	rsdb.SetStatus(req.ID, db.STATUS_QUEUED)
 	rsdb.Claim(req.ID)
 
-	scheduler.AddAt(req.RunAt, func() {
+	reqscheduler.AddAtWithID(req.ID, req.RunAt, func() {
 		HandleScheduledRequest(req)
 	})
 }
 
 func OnRequestScheduled(req *db.ScheduledRequest) {
-	next, exists := scheduler.PeekTime()
+	id, latest, exists := reqscheduler.Latest()
+
+	if !exists {
+		return
+	}
 
-	if exists {
-		if req.RunAt.Before(next) {
-			// add earliest job (current)
-			AddToQueue(req)
+	// get next request
+	nexts, _ := rsdb.FetchNext(1, 0)
 
-			// remove latest job
-			scheduler.Pop()
+	if len(nexts) != 1 {
+		return
+	}
 
-			rsdb.SetStatus(req.ID, db.STATUS_PENDING)
-		}
+	// check if current request runs before the latest heap request
+	// and if the next request in the db is also the current request
+	if req.RunAt.Before(latest) && req.ID == nexts[0].ID {
+		// add current job (earlier than latest)
+		AddToQueue(req)
+
+		// remove latest job
+		reqscheduler.Cancel(id)
+
+		// mark removed request as pending (not queued anymore)
+		rsdb.SetStatus(id, db.STATUS_PENDING)
 	}
 }
 

internals/scheduler/scheduler.go

@@ -1,23 +1,5 @@
 package scheduler
 
-import (
-	"context"
-
-	scheduling "github.com/codeshelldev/gotl/pkg/scheduler"
-)
-
-var scheduler = scheduling.New()
-var cancel context.CancelFunc
-
 func Start() {
-	var ctx context.Context
-	ctx, cancel = context.WithCancel(context.Background())
-
-	go scheduler.Run(ctx)
-
 	StartRequestScheduler()
-}
-
-func Stop() {
-	cancel()
 }