Revert "[ISSUE 246] DST: Fault Injection (#248)" (#251)

This reverts commit 60bda4a.

Author: vaibhawvipul

cmd/dst/run.go

@@ -26,17 +26,16 @@ import (
 
 func RunDSTCmd() *cobra.Command {
 	var (
-		seed               int64
-		ticks              int64
-		reqsPerTick        = util.RangeIntFlag{Min: 1, Max: 1000}
-		ids                = util.RangeIntFlag{Min: 1, Max: 1000}
-		idempotencyKeys    = util.RangeIntFlag{Min: 1, Max: 1000}
-		headers            = util.RangeIntFlag{Min: 1, Max: 1000}
-		data               = util.RangeIntFlag{Min: 1, Max: 1000}
-		tags               = util.RangeIntFlag{Min: 1, Max: 1000}
-		urls               = util.RangeIntFlag{Min: 1, Max: 1000}
-		retries            = util.RangeIntFlag{Min: 1, Max: 1000}
-		failureProbability float64
+		seed            int64
+		ticks           int64
+		reqsPerTick     = util.RangeIntFlag{Min: 1, Max: 1000}
+		ids             = util.RangeIntFlag{Min: 1, Max: 1000}
+		idempotencyKeys = util.RangeIntFlag{Min: 1, Max: 1000}
+		headers         = util.RangeIntFlag{Min: 1, Max: 1000}
+		data            = util.RangeIntFlag{Min: 1, Max: 1000}
+		tags            = util.RangeIntFlag{Min: 1, Max: 1000}
+		urls            = util.RangeIntFlag{Min: 1, Max: 1000}
+		retries         = util.RangeIntFlag{Min: 1, Max: 1000}
 	)
 
 	cmd := &cobra.Command{
@@ -85,7 +84,7 @@ func RunDSTCmd() *cobra.Command {
 
 			// instatiate api/aio
 			api := api.New(config.API.Size, metrics)
-			aio := aio.NewDST(r, metrics, failureProbability)
+			aio := aio.NewDST(r, metrics)
 
 			// instatiate aio subsystems
 			network := network.NewDST(config.AIO.Subsystems.NetworkDST.Config, rand.New(rand.NewSource(r.Int63())))
@@ -207,7 +206,6 @@ func RunDSTCmd() *cobra.Command {
 	cmd.Flags().Var(&tags, "tags", "number promise tags")
 	cmd.Flags().Var(&urls, "urls", "number subscription urls")
 	cmd.Flags().Var(&retries, "retries", "number subscription retries")
-	cmd.Flags().Float64Var(&failureProbability, "failure-probability", 0.5, "probability of aio failure")
 
 	// api
 	cmd.Flags().Var(&util.RangeIntFlag{Min: 1, Max: 1000000}, "api-size", "size of the submission queue buffered channel")

internal/aio/aio_dst.go

@@ -5,7 +5,6 @@ import (
 	"log/slog"
 	"math/rand" // nosemgrep
 
-	"github.com/resonatehq/resonate/internal/kernel/metadata"
 	"github.com/resonatehq/resonate/internal/kernel/t_aio"
 	"github.com/resonatehq/resonate/internal/metrics"
 
@@ -14,20 +13,18 @@ import (
 )
 
 type aioDST struct {
-	r                  *rand.Rand
-	sqes               []*bus.SQE[t_aio.Submission, t_aio.Completion]
-	cqes               []*bus.CQE[t_aio.Submission, t_aio.Completion]
-	subsystems         map[t_aio.Kind]Subsystem
-	metrics            *metrics.Metrics
-	failureProbability float64
+	r          *rand.Rand
+	sqes       []*bus.SQE[t_aio.Submission, t_aio.Completion]
+	cqes       []*bus.CQE[t_aio.Submission, t_aio.Completion]
+	subsystems map[t_aio.Kind]Subsystem
+	metrics    *metrics.Metrics
 }
 
-func NewDST(r *rand.Rand, metrics *metrics.Metrics, failureProbability float64) *aioDST {
+func NewDST(r *rand.Rand, metrics *metrics.Metrics) *aioDST {
 	return &aioDST{
-		r:                  r,
-		subsystems:         map[t_aio.Kind]Subsystem{},
-		metrics:            metrics,
-		failureProbability: failureProbability,
+		r:          r,
+		subsystems: map[t_aio.Kind]Subsystem{},
+		metrics:    metrics,
 	}
 }
 
@@ -70,6 +67,8 @@ func (a *aioDST) Dequeue(n int) []*bus.CQE[t_aio.Submission, t_aio.Completion] {
 	a.cqes = a.cqes[min(n, len(a.cqes)):]
 
 	for _, cqe := range cqes {
+		slog.Debug("aio:dequeue", "cqe", cqe)
+
 		var status string
 		if cqe.Error != nil {
 			status = "failure"
@@ -93,30 +92,7 @@ func (a *aioDST) Flush(t int64) {
 
 	for _, sqes := range util.OrderedRangeKV(flush) {
 		if subsystem, ok := a.subsystems[sqes.Key]; ok {
-			// Randomly decide whether to process SQE or return an error
-			if a.r.Float64() < a.failureProbability {
-				// Create a new CQE
-				errorCqe := createErrorCQE("aio dst error occurred while processing SQE", "aio dst failure", simpleCallback)
-				errorCqe.Completion = &t_aio.Completion{}
-
-				slog.Debug("aio dst: failure", "err", errorCqe)
-				a.cqes = append(a.cqes, errorCqe)
-			} else {
-				// Do the I/O
-				processedCQEs := subsystem.NewWorker(0).Process(sqes.Value)
-				// Randomly decide whether to return an error after processing SQE
-				if a.r.Float64() < a.failureProbability {
-					for _, cqe := range processedCQEs {
-						errorCqe := createErrorCQE("aio dst error occurred after processing SQE", "aio dst failure", simpleCallback)
-						errorCqe.Completion = &t_aio.Completion{}
-
-						slog.Debug("aio dst: failure", "err", errorCqe, "cqe", cqe)
-						a.cqes = append(a.cqes, cqe, errorCqe)
-					}
-				} else {
-					a.cqes = append(a.cqes, processedCQEs...)
-				}
-			}
+			a.cqes = append(a.cqes, subsystem.NewWorker(0).Process(sqes.Value)...)
 		} else {
 			panic("invalid aio submission")
 		}
@@ -125,19 +101,6 @@ func (a *aioDST) Flush(t int64) {
 	a.sqes = nil
 }
 
-func simpleCallback(completion *t_aio.Completion, err error) {
-}
-
-func createErrorCQE(errorMsg string, tags string, callback func(*t_aio.Completion, error)) *bus.CQE[t_aio.Submission, t_aio.Completion] {
-	errorCqe := &bus.CQE[t_aio.Submission, t_aio.Completion]{
-		Error:    fmt.Errorf(errorMsg),
-		Metadata: metadata.New("dst error"),
-	}
-	errorCqe.Metadata.Tags.Set("aio", tags)
-	errorCqe.Callback = callback
-	return errorCqe
-}
-
 func (a *aioDST) String() string {
 	// use subsystem keys so that we can compare cross-store dst runs
 	subsystems := make([]t_aio.Kind, len(a.subsystems))

test/dst/dst_test.go

@@ -36,8 +36,7 @@ func TestDST(t *testing.T) {
 
 	// instatiate api/aio
 	api := api.New(1000, metrics)
-	failureProbability := 0.5
-	aio := aio.NewDST(r, metrics, failureProbability)
+	aio := aio.NewDST(r, metrics)
 
 	// instatiate aio subsystems
 	network := network.NewDST(&network.ConfigDST{P: 0.5}, r)