reset cursors and deduplicate blobs in controller

Author: ian-shim

api/docs/disperser_v2.html

@@ -157,8 +157,6 @@ enum BlobStatus {
   // and in doing so requesting that the validators sign to acknowledge receipt of the blob.
   // Requests that timeout or receive errors are resubmitted to DA nodes for some period of time set by the disperser,
   // after which the BlobStatus becomes COMPLETE.
-  //
-  // Note: this status is not currently implemented, and is a placeholder for future functionality.
   GATHERING_SIGNATURES = 3;
 
   // COMPLETE means the blob has been dispersed to DA nodes, and the GATHERING_SIGNATURES period of time has completed.

api/docs/disperser_v2.md

@@ -56,7 +56,7 @@ type Client interface {
 	UpdateItemWithCondition(ctx context.Context, tableName string, key Key, item Item, condition expression.ConditionBuilder) (Item, error)
 	IncrementBy(ctx context.Context, tableName string, key Key, attr string, value uint64) (Item, error)
 	GetItem(ctx context.Context, tableName string, key Key) (Item, error)
-	GetItems(ctx context.Context, tableName string, keys []Key) ([]Item, error)
+	GetItems(ctx context.Context, tableName string, keys []Key, consistentRead bool) ([]Item, error)
 	QueryIndex(ctx context.Context, tableName string, indexName string, keyCondition string, expAttributeValues ExpressionValues) ([]Item, error)
 	Query(ctx context.Context, tableName string, keyCondition string, expAttributeValues ExpressionValues) ([]Item, error)
 	QueryWithInput(ctx context.Context, input *dynamodb.QueryInput) ([]Item, error)
@@ -276,8 +276,8 @@ func (c *client) GetItem(ctx context.Context, tableName string, key Key) (Item,
 
 // GetItems returns the items for the given keys
 // Note: ordering of items is not guaranteed
-func (c *client) GetItems(ctx context.Context, tableName string, keys []Key) ([]Item, error) {
-	items, err := c.readItems(ctx, tableName, keys)
+func (c *client) GetItems(ctx context.Context, tableName string, keys []Key, consistentRead bool) ([]Item, error) {
+	items, err := c.readItems(ctx, tableName, keys, consistentRead)
 	if err != nil {
 		return nil, err
 	}
@@ -444,7 +444,12 @@ func (c *client) writeItems(ctx context.Context, tableName string, requestItems
 	return failedItems, nil
 }
 
-func (c *client) readItems(ctx context.Context, tableName string, keys []Key) ([]Item, error) {
+func (c *client) readItems(
+	ctx context.Context,
+	tableName string,
+	keys []Key,
+	consistentRead bool,
+) ([]Item, error) {
 	startIndex := 0
 	items := make([]Item, 0)
 	for startIndex < len(keys) {
@@ -454,7 +459,8 @@ func (c *client) readItems(ctx context.Context, tableName string, keys []Key) ([
 		output, err := c.dynamoClient.BatchGetItem(ctx, &dynamodb.BatchGetItemInput{
 			RequestItems: map[string]types.KeysAndAttributes{
 				tableName: {
-					Keys: keysBatch,
+					Keys:           keysBatch,
+					ConsistentRead: aws.Bool(consistentRead),
 				},
 			},
 		})

api/docs/eigenda-protos.html

@@ -296,7 +296,7 @@ func TestBatchOperations(t *testing.T) {
 		}
 	}
 
-	fetchedItems, err := dynamoClient.GetItems(ctx, tableName, keys)
+	fetchedItems, err := dynamoClient.GetItems(ctx, tableName, keys, true)
 	assert.NoError(t, err)
 	assert.Len(t, fetchedItems, numItems)
 	blobKeys := make([]string, numItems)

api/docs/eigenda-protos.md

@@ -59,7 +59,7 @@ func (c *MockDynamoDBClient) GetItem(ctx context.Context, tableName string, key
 	return args.Get(0).(dynamodb.Item), args.Error(1)
 }
 
-func (c *MockDynamoDBClient) GetItems(ctx context.Context, tableName string, keys []dynamodb.Key) ([]dynamodb.Item, error) {
+func (c *MockDynamoDBClient) GetItems(ctx context.Context, tableName string, keys []dynamodb.Key, consistentRead bool) ([]dynamodb.Item, error) {
 	args := c.Called()
 	return args.Get(0).([]dynamodb.Item), args.Error(1)
 }

api/grpc/disperser/v2/disperser_v2.pb.go

@@ -350,14 +350,22 @@ func (a *Attestation) ToProtobuf() (*disperserpb.Attestation, error) {
 		quorumResults[i] = a.QuorumResults[q]
 	}
 
-	apkG2Bytes := a.APKG2.Bytes()
-	sigmaBytes := a.Sigma.Bytes()
+	var apkG2Bytes []byte
+	var sigmaBytes []byte
+	if a.APKG2 != nil {
+		b := a.APKG2.Bytes()
+		apkG2Bytes = b[:]
+	}
+	if a.Sigma != nil {
+		b := a.Sigma.Bytes()
+		sigmaBytes = b[:]
+	}
 
 	return &disperserpb.Attestation{
 		NonSignerPubkeys:        nonSignerPubKeys,
-		ApkG2:                   apkG2Bytes[:],
+		ApkG2:                   apkG2Bytes,
 		QuorumApks:              quorumAPKs,
-		Sigma:                   sigmaBytes[:],
+		Sigma:                   sigmaBytes,
 		QuorumNumbers:           quorumNumbers,
 		QuorumSignedPercentages: quorumResults,
 	}, nil