feat: cas scaling

common/aws/ddb/job.go

@@ -20,6 +20,8 @@ import (
 	"github.com/ceramicnetwork/go-cas/models"
 )
 
+var _ models.JobRepository = &JobDatabase{}
+
 type JobDatabase struct {
 	ddbClient *dynamodb.Client
 	table     string
@@ -43,17 +45,15 @@ func (jdb *JobDatabase) createJobTable(ctx context.Context) error {
 
 func (jdb *JobDatabase) CreateJob(ctx context.Context) (string, error) {
 	jobParams := map[string]interface{}{
-		job.JobParam_Version:   models.WorkerVersion, // this will launch a CASv5 Worker
-		job.JobParam_Overrides: map[string]string{},
-	}
-	// Only enable continuous anchoring on Clay and Prod
-	if (jdb.env == cas.EnvTag_Tnet) || (jdb.env == cas.EnvTag_Prod) {
-		jobParams[job.JobParam_Overrides].(map[string]string)[models.AnchorOverrides_AppMode] = models.AnchorAppMode_ContinualAnchoring
-		jobParams[job.JobParam_Overrides].(map[string]string)[models.AnchorOverrides_SchedulerStopAfterNoOp] = "true"
+		job.AnchorJobParam_Version: models.WorkerVersion, // this will launch a CASv5 Worker
+		job.AnchorJobParam_Overrides: map[string]string{
+			models.AnchorOverrides_AppMode:                models.AnchorAppMode_ContinualAnchoring,
+			models.AnchorOverrides_SchedulerStopAfterNoOp: "true",
+		},
 	}
 	// If an override anchor contract address is available, pass it through to the job.
 	if contractAddress, found := os.LookupEnv(models.Env_AnchorContractAddress); found {
-		jobParams[job.JobParam_Overrides].(map[string]string)[models.AnchorOverrides_ContractAddress] = contractAddress
+		jobParams[job.AnchorJobParam_Overrides].(map[string]string)[models.AnchorOverrides_ContractAddress] = contractAddress
 	}
 	newJob := models.NewJob(job.JobType_Anchor, jobParams)
 	attributeValues, err := attributevalue.MarshalMapWithOptions(newJob, func(options *attributevalue.EncoderOptions) {
@@ -74,7 +74,7 @@ func (jdb *JobDatabase) CreateJob(ctx context.Context) (string, error) {
 		if err != nil {
 			return "", err
 		} else {
-			return newJob.Job, nil
+			return newJob.JobId, nil
 		}
 	}
 }

common/aws/ddb/state.go

@@ -19,6 +19,8 @@ import (
 	"github.com/ceramicnetwork/go-cas/models"
 )
 
+var _ models.StateRepository = &StateDatabase{}
+
 type StateDatabase struct {
 	client          *dynamodb.Client
 	checkpointTable string

common/aws/queue/monitor.go

@@ -8,6 +8,8 @@ import (
 	"github.com/ceramicnetwork/go-cas/models"
 )
 
+var _ models.QueueMonitor = &Monitor{}
+
 type Monitor struct {
 	queueUrl string
 	client   *sqs.Client

common/aws/queue/multiMonitor.go

@@ -0,0 +1,32 @@
+package queue
+
+import (
+	"context"
+
+	"github.com/ceramicnetwork/go-cas/models"
+)
+
+var _ models.QueueMonitor = &MultiMonitor{}
+
+type MultiMonitor struct {
+	monitors []models.QueueMonitor
+}
+
+func NewMultiMonitor(monitors []models.QueueMonitor) models.QueueMonitor {
+	return &MultiMonitor{monitors}
+}
+
+func (m MultiMonitor) GetUtilization(ctx context.Context) (int, int, error) {
+	// Get the utilization of all sub-queues
+	totalMsgsUnprocessed := 0
+	totalMsgsInFlight := 0
+	for _, monitor := range m.monitors {
+		numMsgsUnprocessed, numMsgsInFlight, err := monitor.GetUtilization(ctx)
+		if err != nil {
+			return 0, 0, err
+		}
+		totalMsgsUnprocessed += numMsgsUnprocessed
+		totalMsgsInFlight += numMsgsInFlight
+	}
+	return totalMsgsUnprocessed, totalMsgsInFlight, nil
+}

common/aws/queue/multiQueue.go

@@ -0,0 +1,86 @@
+package queue
+
+import (
+	"context"
+	"math"
+	"sync/atomic"
+
+	"github.com/aws/aws-sdk-go-v2/service/sqs"
+
+	"github.com/abevier/go-sqs/gosqs"
+
+	"github.com/ceramicnetwork/go-cas/models"
+)
+
+var _ models.Queue = &multiQueue{}
+var _ models.QueuePublisher = &multiQueue{}
+
+type multiQueue struct {
+	queues  []models.Queue
+	monitor models.QueueMonitor
+	index   int32
+}
+
+func NewMultiQueue(
+	ctx context.Context,
+	metricService models.MetricService,
+	logger models.Logger,
+	sqsClient *sqs.Client,
+	opts Opts,
+	callback gosqs.MessageCallbackFunc,
+	numQueues int,
+) (models.Queue, error) {
+	// Create at least one sub-queue
+	numQueues = int(math.Max(1, float64(numQueues)))
+	queues := make([]models.Queue, numQueues)
+	monitors := make([]models.QueueMonitor, numQueues)
+	if opts.NumWorkers != nil {
+		// Divide the workers evenly among the sub-queues
+		*opts.NumWorkers = int(math.Ceil(float64(*opts.NumWorkers / numQueues)))
+	}
+	for i := 0; i < numQueues; i++ {
+		if q, _, err := newQueue(ctx, metricService, logger, sqsClient, opts, callback, i); err != nil {
+			return nil, err
+		} else {
+			queues[i] = q
+			monitors[i] = q.Monitor()
+		}
+	}
+	return &multiQueue{queues, NewMultiMonitor(monitors), 0}, nil
+}
+
+func (m *multiQueue) SendMessage(ctx context.Context, event any) (string, error) {
+	// Use an atomic counter to round-robin between sub-queues
+	i := atomic.AddInt32(&m.index, 1) - 1
+	i = i % int32(len(m.queues))
+	return m.queues[i].Publisher().SendMessage(ctx, event)
+}
+
+func (m *multiQueue) Start() {
+	// Start all sub-queues
+	for _, q := range m.queues {
+		q.Start()
+	}
+}
+
+func (m *multiQueue) Shutdown() {
+	// Shutdown all sub-queues
+	for _, q := range m.queues {
+		q.Shutdown()
+	}
+}
+
+func (m *multiQueue) WaitForRxShutdown() {
+	// Wait for all sub-queues to shutdown
+	for _, q := range m.queues {
+		q.WaitForRxShutdown()
+	}
+}
+
+func (m *multiQueue) Monitor() models.QueueMonitor {
+	return m.monitor
+}
+
+func (m *multiQueue) Publisher() models.QueuePublisher {
+	return m
+}