feat(back-channel): verify the task submissions

cmd/dst/run.go

@@ -113,6 +113,9 @@ func RunDSTCmd() *cobra.Command {
 			if err != nil {
 				return err
 			}
+
+			reqsPerTickSize := reqsPerTick.Resolve(r)
+			config.AIO.Subsystems.QueuingDST.Config.Queue = make(chan *queuing.ConnectionSubmissionDST, reqsPerTickSize)
 			queuing, err := queuing.NewDST(config.AIO.Subsystems.QueuingDST.Config, rand.New(rand.NewSource(r.Int63())))
 			if err != nil {
 				return err
@@ -191,10 +194,11 @@ func RunDSTCmd() *cobra.Command {
 
 			dst := dst.New(&dst.Config{
 				Scenario:           dstScenario,
+				TaskQueue:          config.AIO.Subsystems.QueuingDST.Config.Queue,
 				Ticks:              ticks,
 				TimeElapsedPerTick: 50_000, // milliseconds
 				Reqs: func() int {
-					return reqsPerTick.Resolve(r)
+					return reqsPerTickSize
 				},
 				Ids:             ids.Resolve(r),
 				IdempotencyKeys: idempotencyKeys.Resolve(r),

internal/app/subsystems/aio/queuing/queuing_dst.go

@@ -11,7 +11,8 @@ import (
 
 type (
 	ConfigDST struct {
-		P float32
+		P     float32
+		Queue chan *ConnectionSubmissionDST
 	}
 
 	QueuingSubsystemDST struct {
@@ -23,6 +24,12 @@ type (
 		config *ConfigDST
 		r      *rand.Rand
 	}
+
+	ConnectionSubmissionDST struct {
+		Queue   string `json:"queue"`
+		TaskId  string `json:"taskid"`
+		Counter int    `json:"counter"`
+	}
 )
 
 // NewDST is a simple helper functions that wraps New and returns a pre-configured QueuingSubsystem.
@@ -49,6 +56,7 @@ func (q *QueuingSubsystemDST) Start() error {
 }
 
 func (q *QueuingSubsystemDST) Stop() error {
+	close(q.config.Queue)
 	return nil
 }
 
@@ -69,6 +77,12 @@ func (w *QueuingWorkerDST) Process(sqes []*bus.SQE[t_aio.Submission, t_aio.Compl
 	for i, sqe := range sqes {
 		util.Assert(sqe.Submission.Queuing != nil, "submission must not be nil")
 
+		w.config.Queue <- &ConnectionSubmissionDST{
+			Queue:   "analytics",
+			TaskId:  sqe.Submission.Queuing.TaskId,
+			Counter: sqe.Submission.Queuing.Counter,
+		}
+
 		cqe := &bus.CQE[t_aio.Submission, t_aio.Completion]{
 			Metadata: sqe.Metadata,
 			Callback: sqe.Callback,

test/dst/dst.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/resonatehq/resonate/internal/aio"
 	"github.com/resonatehq/resonate/internal/api"
+	"github.com/resonatehq/resonate/internal/app/subsystems/aio/queuing"
 	"github.com/resonatehq/resonate/internal/kernel/bus"
 	"github.com/resonatehq/resonate/internal/kernel/metadata"
 	"github.com/resonatehq/resonate/internal/kernel/system"
@@ -20,6 +21,7 @@ type DST struct {
 
 type Config struct {
 	Scenario           *Scenario
+	TaskQueue          chan *queuing.ConnectionSubmissionDST
 	Ticks              int64
 	TimeElapsedPerTick int64
 	Reqs               func() int
@@ -66,7 +68,7 @@ func (d *DST) Run(r *rand.Rand, api api.API, aio aio.AIO, system *system.System,
 	generator := NewGenerator(r, d.config)
 
 	// model
-	model := NewModel(d.config.Scenario)
+	model := NewModel(d.config.Scenario, d.config.TaskQueue)
 
 	// add req/res
 	for _, req := range reqs {

test/dst/dst_test.go

@@ -69,7 +69,10 @@ func test(t *testing.T, scenario *Scenario) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	queuing, err := queuing.NewDST(&queuing.ConfigDST{P: 0.5}, r)
+
+	reqsPerTickSize := 100
+	taskQueue := make(chan *queuing.ConnectionSubmissionDST, reqsPerTickSize)
+	queuing, err := queuing.NewDST(&queuing.ConfigDST{P: 0.5, Queue: taskQueue}, r)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -147,9 +150,9 @@ func test(t *testing.T, scenario *Scenario) {
 	}
 
 	dst := New(&Config{
-		Ticks:              1000,
+		Ticks:              3_000,
 		TimeElapsedPerTick: 50_000, // milliseconds
-		Reqs:               func() int { return 100 },
+		Reqs:               func() int { return reqsPerTickSize },
 		Ids:                100,
 		IdempotencyKeys:    100,
 		Headers:            100,
@@ -158,6 +161,7 @@ func test(t *testing.T, scenario *Scenario) {
 		Urls:               100,
 		Retries:            100,
 		Scenario:           scenario,
+		TaskQueue:          taskQueue,
 	})
 
 	if errs := dst.Run(r, api, aio, system, reqs); len(errs) > 0 {

test/dst/model.go

@@ -5,6 +5,7 @@
 import (
 	"errors"
 	"fmt"
+	"log/slog"
 	"regexp"
 	"strings"
 
@@ -21,13 +22,15 @@
 // Model
 
 type Model struct {
-	scenario  *Scenario
-	promises  Promises
-	schedules Schedules
-	locks     Locks
-	tasks     Tasks
-	cursors   []*t_api.Request
-	responses map[t_api.Kind]ResponseValidator
+	scenario             *Scenario
+	promises             Promises
+	schedules            Schedules
+	locks                Locks
+	tasks                Tasks
+	taskQueueSubmissions TaskQueueSubmissions
+	cursors              []*t_api.Request
+	responses            map[t_api.Kind]ResponseValidator
+	taskQueue            chan *queuing.ConnectionSubmissionDST
 }
 
 type PromiseModel struct {
@@ -56,11 +59,17 @@
 	task *task.Task
 }
 
+type TaskQueueSubmissionModel struct {
+	id                  string
+	taskQueueSubmission *queuing.ConnectionSubmissionDST
+}
+
 type Promises map[string]*PromiseModel
 type Schedules map[string]*ScheduleModel
 type Subscriptions map[string]*SubscriptionModel
 type Locks map[string]*LockModel
 type Tasks map[string]*TaskModel
+type TaskQueueSubmissions map[string]*TaskQueueSubmissionModel
 type ResponseValidator func(int64, int64, *t_api.Request, *t_api.Response) error
 
 func (p Promises) Get(id string) *PromiseModel {
@@ -114,14 +123,26 @@
 	return t[id]
 }
 
-func NewModel(scenario *Scenario) *Model {
+func (t TaskQueueSubmissions) Get(id string) *TaskQueueSubmissionModel {
+	if _, ok := t[id]; !ok {
+		t[id] = &TaskQueueSubmissionModel{
+			id: id,
+		}
+	}
+
+	return t[id]
+}
+
+func NewModel(scenario *Scenario, queue chan *queuing.ConnectionSubmissionDST) *Model {
 	return &Model{
-		scenario:  scenario,
-		promises:  map[string]*PromiseModel{},
-		schedules: map[string]*ScheduleModel{},
-		locks:     map[string]*LockModel{},
-		tasks:     map[string]*TaskModel{},
-		responses: map[t_api.Kind]ResponseValidator{},
+		scenario:             scenario,
+		promises:             map[string]*PromiseModel{},
+		schedules:            map[string]*ScheduleModel{},
+		locks:                map[string]*LockModel{},
+		tasks:                map[string]*TaskModel{},
+		taskQueueSubmissions: map[string]*TaskQueueSubmissionModel{},
+		responses:            map[t_api.Kind]ResponseValidator{},
+		taskQueue:            queue,
 	}
 }
 
@@ -161,13 +182,48 @@
 		return fmt.Errorf("unexpected response kind '%d' for request kind '%d'", res.Kind, req.Kind)
 	}
 
+	if err := m.dequeue(); err != nil {
+		return err
+	}
+
 	if f, ok := m.responses[req.Kind]; ok {
 		return f(reqTime, resTime, req, res)
 	}
 
 	return nil
 }
 
+func (m *Model) dequeue() error {
+	for {
+		select {
+		case task, ok := <-m.taskQueue:
+			if !ok {
+				// queue closed
+				return nil
+			}
+
+			slog.Info("dequeue", "task", task)
+
+			submission := m.taskQueueSubmissions.Get(task.TaskId)
+			if submission.taskQueueSubmission == nil {
+				if task.Counter != 1 {
+					return fmt.Errorf("unexpected counter %d, expected 1", task.Counter)
+				}
+				m.taskQueueSubmissions[task.TaskId] = &TaskQueueSubmissionModel{
+					id:                  task.TaskId,
+					taskQueueSubmission: task,
+				}
+				continue
+			}
+			if submission.taskQueueSubmission.Counter >= task.Counter {
+				return fmt.Errorf("unexpected counter %d, expected %d", submission.taskQueueSubmission.Counter, task.Counter)
+			}
+		default: // no more requests
+			return nil
+		}
+	}
+}
+
 // PROMISES
 
 func (m *Model) ValidateReadPromise(reqTime int64, resTime int64, req *t_api.Request, res *t_api.Response) error {