.gitea/workflows/deploy-dev.yaml

@@ -9,7 +9,7 @@ on:
 jobs:
   build:
     name: Build
-    runs-on: [dev]
+    runs-on: [ dev ]
     steps:
       - name: login
         run: docker login -u mathwave -p ${{ secrets.DOCKERHUB_PASSWORD }}
@@ -21,11 +21,18 @@ jobs:
         run: docker build -t mathwave/sprint-repo:queues .
   push:
     name: Push
-    runs-on: [dev]
+    runs-on: [ dev ]
     needs: build
     steps:
       - name: push
         run: docker push mathwave/sprint-repo:queues
+  create_dir:
+    name: Create dir
+    runs-on: [ dev ]
+    needs: build
+    steps:
+      - name: create_dir
+        run: mkdir /sprint-data/queues-mongo || true
   deploy-dev:
     name: Deploy dev
     runs-on: [prod]

.gitea/workflows/deploy-prod.yaml

@@ -9,7 +9,7 @@ on:
 jobs:
   build:
     name: Build
-    runs-on: [dev]
+    runs-on: [ dev ]
     steps:
       - name: login
         run: docker login -u mathwave -p ${{ secrets.DOCKERHUB_PASSWORD }}
@@ -21,11 +21,18 @@ jobs:
         run: docker build -t mathwave/sprint-repo:queues .
   push:
     name: Push
-    runs-on: [dev]
+    runs-on: [ dev ]
     needs: build
     steps:
       - name: push
         run: docker push mathwave/sprint-repo:queues
+  create_dir:
+    name: Create dir
+    runs-on: [ prod ]
+    needs: build
+    steps:
+      - name: create_dir
+        run: mkdir /sprint-data/queues-mongo || true
   deploy-prod:
     name: Deploy prod
     runs-on: [prod]

app/storage/mongo/collections/tasks.go

@@ -9,7 +9,6 @@ import (
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/bson/primitive"
 	"go.mongodb.org/mongo-driver/mongo"
-	"go.mongodb.org/mongo-driver/mongo/options"
 )
 
 type Task struct {
@@ -63,19 +62,7 @@ func Take(queue string) (*Task, error) {
 	if task == nil {
 		return nil, nil
 	}
-	_, err = collection().UpdateByID(
-		context.TODO(),
-		task.Id,
-		bson.M{
-			"$set": bson.M{
-				"taken_at": now,
-				"attempts": task.Attempts + 1,
-				"available_from": now.Add(
-					time.Duration(task.SecondsToExecute) * time.Second,
-				),
-			},
-		},
-	)
+	_, err = collection().UpdateByID(context.TODO(), task.Id, bson.M{"$set": bson.M{"taken_at": now, "attempts": task.Attempts + 1}})
 	if err != nil {
 		println("Error updaing")
 		println(err.Error())
@@ -91,7 +78,6 @@ func findTask(queue string, now time.Time) (*Task, error) {
 			"queue":          queue,
 			"available_from": bson.M{"$lte": now},
 		},
-		options.Find().SetLimit(1),
 	)
 	if err != nil {
 		println("Error find")
@@ -108,7 +94,14 @@ func findTask(queue string, now time.Time) (*Task, error) {
 	}
 
 	for _, task := range results {
-		return &task, nil
+		if task.TakenAt == nil {
+			return &task, nil
+		}
+		takenAt := *task.TakenAt
+
+		if takenAt.Add(time.Second * time.Duration(task.SecondsToExecute)).Before(now) {
+			return &task, nil
+		}
 	}
 	return nil, nil
 }