Add strict mode and remove complete promise

cmd/dst.go

@@ -118,7 +118,7 @@ var dstRunCmd = &cobra.Command{
 
 		// instatiate api/aio
 		api := api.New(config.API.Size, metrics)
-		aio := aio.NewDST()
+		aio := aio.NewDST(r)
 
 		// instatiate aio subsystems
 		network := network.NewDST(config.AIO.Subsystems.NetworkDST.Config, rand.New(rand.NewSource(r.Int63())))

cmd/serve.go

@@ -83,7 +83,6 @@ var serveCmd = &cobra.Command{
 		system.AddOnRequest(types.ResolvePromise, coroutines.ResolvePromise)
 		system.AddOnRequest(types.RejectPromise, coroutines.RejectPromise)
 		system.AddOnRequest(types.CancelPromise, coroutines.CancelPromise)
-		system.AddOnRequest(types.CompletePromise, coroutines.CompletePromise)
 		system.AddOnRequest(types.ReadSubscriptions, coroutines.ReadSubscriptions)
 		system.AddOnRequest(types.CreateSubscription, coroutines.CreateSubscription)
 		system.AddOnRequest(types.DeleteSubscription, coroutines.DeleteSubscription)

internal/aio/aio_dst.go

@@ -2,21 +2,24 @@ package aio
 
 import (
 	"fmt"
+	"math/rand" // nosemgrep
 
 	"github.com/resonatehq/resonate/internal/kernel/bus"
 	"github.com/resonatehq/resonate/internal/kernel/types"
 	"github.com/resonatehq/resonate/internal/util"
 )
 
 type aioDST struct {
+	r          *rand.Rand
 	sqes       []*bus.SQE[types.Submission, types.Completion]
 	cqes       []*bus.CQE[types.Submission, types.Completion]
 	subsystems map[types.AIOKind]Subsystem
 	done       bool
 }
 
-func NewDST() *aioDST {
+func NewDST(r *rand.Rand) *aioDST {
 	return &aioDST{
+		r:          r,
 		subsystems: map[types.AIOKind]Subsystem{},
 	}
 }
@@ -54,7 +57,8 @@ func (a *aioDST) Done() bool {
 }
 
 func (a *aioDST) Enqueue(sqe *bus.SQE[types.Submission, types.Completion]) {
-	a.sqes = append(a.sqes, sqe)
+	i := a.r.Intn(len(a.sqes) + 1)
+	a.sqes = append(a.sqes[:i], append([]*bus.SQE[types.Submission, types.Completion]{sqe}, a.sqes[i:]...)...)
 }
 
 func (a *aioDST) Dequeue(n int) []*bus.CQE[types.Submission, types.Completion] {

internal/app/coroutines/cancelPromise.go

@@ -37,7 +37,7 @@
 
 		c.Yield(submission, func(t int64, completion *types.Completion, err error) {
 			if err != nil {
-				slog.Error("failed to read promise or read subscriptions", "req", req, "err", err)
+				slog.Error("failed to read promise", "req", req, "err", err)
 				res(t, nil, err)
 				return
 			}
@@ -160,11 +160,11 @@
 						})
 					}
 				} else {
-					var status types.ResponseStatus
-					if p.State == promise.Canceled && p.Value.Ikey.Match(req.CancelPromise.Value.Ikey) {
+					status := types.ResponseForbidden
+					strict := req.CancelPromise.Strict && p.State != promise.Canceled
+
+					if !strict && p.Value.Ikey.Match(req.CancelPromise.Value.Ikey) {
 						status = types.ResponseOK
-					} else {
-						status = types.ResponseForbidden
 					}
 
 					res(t, &types.Response{

internal/app/coroutines/createPromise.go

@@ -110,11 +110,11 @@ func CreatePromise(t int64, req *types.Request, res func(int64, *types.Response,
 					return
 				}
 
-				var status types.ResponseStatus
-				if p.Param.Ikey.Match(req.CreatePromise.Param.Ikey) {
+				status := types.ResponseForbidden
+				strict := req.CreatePromise.Strict && p.State != promise.Pending
+
+				if !strict && p.Param.Ikey.Match(req.CreatePromise.Param.Ikey) {
 					status = types.ResponseOK
-				} else {
-					status = types.ResponseForbidden
 				}
 
 				if p.State == promise.Pending && t >= p.Timeout {

internal/app/coroutines/rejectPromise.go

@@ -37,7 +37,7 @@
 
 		c.Yield(submission, func(t int64, completion *types.Completion, err error) {
 			if err != nil {
-				slog.Error("failed to read promise or read subscriptions", "req", req, "err", err)
+				slog.Error("failed to read promise", "req", req, "err", err)
 				res(t, nil, err)
 				return
 			}
@@ -160,11 +160,11 @@
 						})
 					}
 				} else {
-					var status types.ResponseStatus
-					if p.State == promise.Rejected && p.Value.Ikey.Match(req.RejectPromise.Value.Ikey) {
+					status := types.ResponseForbidden
+					strict := req.RejectPromise.Strict && p.State != promise.Rejected
+
+					if !strict && p.Value.Ikey.Match(req.RejectPromise.Value.Ikey) {
 						status = types.ResponseOK
-					} else {
-						status = types.ResponseForbidden
 					}
 
 					res(t, &types.Response{

internal/app/coroutines/resolvePromise.go

@@ -37,7 +37,7 @@
 
 		c.Yield(submission, func(t int64, completion *types.Completion, err error) {
 			if err != nil {
-				slog.Error("failed to read promise or read subscriptions", "req", req, "err", err)
+				slog.Error("failed to read promise", "req", req, "err", err)
 				res(t, nil, err)
 				return
 			}
@@ -159,11 +159,11 @@
 						})
 					}
 				} else {
-					var status types.ResponseStatus
-					if p.State == promise.Resolved && p.Value.Ikey.Match(req.ResolvePromise.Value.Ikey) {
+					status := types.ResponseForbidden
+					strict := req.ResolvePromise.Strict && p.State != promise.Resolved
+
+					if !strict && p.Value.Ikey.Match(req.ResolvePromise.Value.Ikey) {
 						status = types.ResponseOK
-					} else {
-						status = types.ResponseForbidden
 					}
 
 					res(t, &types.Response{