fix: delete blobs

disperser/apiserver/server.go

@@ -243,19 +243,18 @@ func (s *DispersalServer) GetBlobStatus(ctx context.Context, req *pb.BlobStatusR
 		return nil, err
 	}
 
-	s.logger.Debug("metadataKey", "metadataKey", metadataKey.String())
 	metadata, err := s.blobStore.GetBlobMetadata(ctx, metadataKey)
 	if err != nil {
-		if !s.metadataHashAsBlobKey && s.KVNode != nil {
-			return nil, err
-		}
+		return nil, err
+	}
+	if metadata.GetBlobKey().String() != string(requestID) && s.metadataHashAsBlobKey {
 		// check on kv
-		metadata, err = s.getMetadataFromKv(requestID)
+		metadataInKV, err := s.getMetadataFromKv(requestID)
 		if err != nil {
 			s.logger.Warn("get metadata from kv", err)
 		}
-		if metadata == nil {
-			return nil, fmt.Errorf("request not found")
+		if metadataInKV != nil {
+			metadata = metadataInKV
 		}
 	}
 

disperser/batcher/batcher.go

@@ -120,7 +120,9 @@ func (b *Batcher) Start(ctx context.Context) error {
 	b.confirmer.EncodingStreamer = b.EncodingStreamer
 	b.confirmer.Start(ctx)
 	// finalizer
-	b.finalizer.Start(ctx)
+	if !b.Queue.MetadataHashAsBlobKey() {
+		b.finalizer.Start(ctx)
+	}
 
 	go func() {
 		ticker := time.NewTicker(b.PullInterval)

disperser/batcher/confirmer.go

@@ -229,9 +229,15 @@ func (c *Confirmer) PersistConfirmedBlobs(ctx context.Context, metadatas []*disp
 	if err != nil {
 		return errors.WithMessage(err, "failed to confirm metadata onchain")
 	}
+	c.logger.Info("[confirmer] removing confirmed blobs")
 	for _, metadata := range metadatas {
-		c.Queue.RemoveBlob(ctx, metadata)
+		c.logger.Info("[confirmer] removing blob", "blob key", metadata.GetBlobKey().String())
+		err := c.Queue.RemoveBlob(ctx, metadata)
+		if err != nil {
+			c.logger.Warn("[confirmer] failed to remove blob", "error", err)
+		}
 	}
+	c.logger.Info("[confirmer] confirmed blobs removed")
 	return nil
 }
 
@@ -295,10 +301,13 @@ func (c *Confirmer) ConfirmBatch(ctx context.Context, batchInfo *BatchInfo) erro
 
 	// remove blobs
 	if c.Queue.MetadataHashAsBlobKey() {
+		stageTimer = time.Now()
+		c.logger.Info("[confirmer] Uploading confirmed metadata on chain")
 		err := c.PersistConfirmedBlobs(ctx, confirmedMetadatas)
 		if err != nil {
-			c.logger.Error("failed to upload metadata on chain: %v", err)
+			c.logger.Error("[confirmer] Failed to upload metadata on chain: %v", err)
 		}
+		c.logger.Info("[confirmer] Uploaded confirmed metadata on chain", "duration", time.Since(stageTimer))
 	}
 
 	batchSize := int64(0)

disperser/batcher/transactor/transactor.go

@@ -13,7 +13,7 @@ import (
 )
 
 type Transactor struct {
-	mu *sync.Mutex
+	mu sync.Mutex
 
 	logger common.Logger
 }

disperser/cmd/apiserver/flags/flags.go

@@ -4,6 +4,7 @@ import (
 	"github.com/urfave/cli"
 	"github.com/zero-gravity-labs/zerog-data-avail/common"
 	"github.com/zero-gravity-labs/zerog-data-avail/common/aws"
+	"github.com/zero-gravity-labs/zerog-data-avail/common/geth"
 	"github.com/zero-gravity-labs/zerog-data-avail/common/logging"
 	"github.com/zero-gravity-labs/zerog-data-avail/common/ratelimit"
 	"github.com/zero-gravity-labs/zerog-data-avail/common/storage_node"
@@ -96,5 +97,6 @@ func init() {
 	Flags = append(Flags, logging.CLIFlags(envVarPrefix, FlagPrefix)...)
 	Flags = append(Flags, ratelimit.RatelimiterCLIFlags(envVarPrefix, FlagPrefix)...)
 	Flags = append(Flags, aws.ClientFlags(envVarPrefix, FlagPrefix)...)
+	Flags = append(Flags, geth.EthClientFlags(envVarPrefix)...)
 	Flags = append(Flags, storage_node.ClientFlags(envVarPrefix, FlagPrefix)...)
 }

go.mod

@@ -138,7 +138,7 @@ require (
 	github.com/go-stack/stack v1.8.1 // indirect
 	github.com/golang/snappy v0.0.5-0.20220116011046-fa5810519dcb // indirect
 	github.com/google/go-cmp v0.5.9 // indirect
-	github.com/google/uuid v1.3.1
+	github.com/google/uuid v1.3.1 // indirect
 	github.com/gorilla/websocket v1.5.0 // indirect
 	github.com/hashicorp/golang-lru/v2 v2.0.7
 	github.com/holiman/bloomfilter/v2 v2.0.3 // indirect