chore: allow validation service to process subset of formats

archiver/service/archiver.go

@@ -199,7 +199,7 @@ func (a *Archiver) backfillBlobs(ctx context.Context, latest *v1.BeaconBlockHead
 		a.log.Crit("failed to read backfill_processes", "err", err)
 	}
 	backfillProcesses[common.Hash(latest.Root)] = storage.BackfillProcess{Start: *latest, Current: *latest}
-	a.dataStoreClient.WriteBackfillProcesses(ctx, backfillProcesses)
+	_ = a.dataStoreClient.WriteBackfillProcesses(ctx, backfillProcesses)
 
 	backfillLoop := func(start *v1.BeaconBlockHeader, current *v1.BeaconBlockHeader) {
 		curr, alreadyExists, err := current, false, error(nil)
@@ -219,7 +219,7 @@ func (a *Archiver) backfillBlobs(ctx context.Context, latest *v1.BeaconBlockHead
 				"startSlot", start.Header.Message.Slot,
 			)
 			delete(backfillProcesses, common.Hash(start.Root))
-			a.dataStoreClient.WriteBackfillProcesses(ctx, backfillProcesses)
+			_ = a.dataStoreClient.WriteBackfillProcesses(ctx, backfillProcesses)
 		}()
 
 		for !alreadyExists {
@@ -246,7 +246,7 @@ func (a *Archiver) backfillBlobs(ctx context.Context, latest *v1.BeaconBlockHead
 			count++
 			if count%10 == 0 {
 				backfillProcesses[common.Hash(start.Root)] = storage.BackfillProcess{Start: *start, Current: *curr}
-				a.dataStoreClient.WriteBackfillProcesses(ctx, backfillProcesses)
+				_ = a.dataStoreClient.WriteBackfillProcesses(ctx, backfillProcesses)
 			}
 		}
 	}

validator/cmd/main.go

@@ -59,6 +59,6 @@ func Main() cliapp.LifecycleAction {
 		beaconClient := service.NewBlobSidecarClient(cfg.BeaconConfig.BeaconURL)
 		blobClient := service.NewBlobSidecarClient(cfg.BlobConfig.BeaconURL)
 
-		return service.NewValidator(l, headerClient, beaconClient, blobClient, closeApp, cfg.NumBlocks), nil
+		return service.NewValidator(l, headerClient, beaconClient, blobClient, closeApp, cfg), nil
 	}
 }

validator/flags/config.go

@@ -10,10 +10,11 @@ import (
 )
 
 type ValidatorConfig struct {
-	LogConfig    oplog.CLIConfig
-	BeaconConfig common.BeaconConfig
-	BlobConfig   common.BeaconConfig
-	NumBlocks    int
+	LogConfig       oplog.CLIConfig
+	BeaconConfig    common.BeaconConfig
+	BlobConfig      common.BeaconConfig
+	ValidateFormats []string
+	NumBlocks       int
 }
 
 func (c ValidatorConfig) Check() error {
@@ -29,6 +30,21 @@ func (c ValidatorConfig) Check() error {
 		return fmt.Errorf("number of blocks must be greater than 0")
 	}
 
+	if len(c.ValidateFormats) == 0 || len(c.ValidateFormats) > 2 {
+		return fmt.Errorf("no formats to validate, please specify formats [json,ssz]")
+	}
+
+	seen := make(map[string]struct{})
+	for _, format := range c.ValidateFormats {
+		if format != "json" && format != "ssz" {
+			return fmt.Errorf("invalid format %v, please specify formats [json,ssz]", format)
+		}
+		if _, ok := seen[format]; ok {
+			return fmt.Errorf("duplicate format %v", format)
+		}
+		seen[format] = struct{}{}
+	}
+
 	return nil
 }
 
@@ -45,6 +61,7 @@ func ReadConfig(cliCtx *cli.Context) ValidatorConfig {
 			BeaconURL:           cliCtx.String(BlobApiClientUrlFlag.Name),
 			BeaconClientTimeout: timeout,
 		},
-		NumBlocks: cliCtx.Int(NumBlocksClientFlag.Name),
+		NumBlocks:       cliCtx.Int(NumBlocksClientFlag.Name),
+		ValidateFormats: cliCtx.StringSlice(ValidateFormatsFlag.Name),
 	}
 }

validator/flags/flags.go

@@ -34,11 +34,18 @@ var (
 		Required: true,
 		EnvVars:  opservice.PrefixEnvVar(EnvVarPrefix, "NUM_BLOCKS"),
 	}
+	ValidateFormatsFlag = &cli.StringSliceFlag{
+		Name:     "validate-formats",
+		Usage:    "The formats to validate [json, ssz]",
+		Value:    cli.NewStringSlice("json", "ssz"),
+		Required: true,
+		EnvVars:  opservice.PrefixEnvVar(EnvVarPrefix, "VALIDATE_FORMATS"),
+	}
 )
 
 func init() {
 	Flags = append(Flags, oplog.CLIFlags(EnvVarPrefix)...)
-	Flags = append(Flags, BeaconClientTimeoutFlag, L1BeaconClientUrlFlag, BlobApiClientUrlFlag, NumBlocksClientFlag)
+	Flags = append(Flags, BeaconClientTimeoutFlag, L1BeaconClientUrlFlag, BlobApiClientUrlFlag, NumBlocksClientFlag, ValidateFormatsFlag)
 }
 
 // Flags contains the list of configuration options available to the binary.

validator/service/service.go

@@ -14,6 +14,7 @@ import (
 	v1 "github.com/attestantio/go-eth2-client/api/v1"
 	"github.com/attestantio/go-eth2-client/spec/phase0"
 	"github.com/base-org/blob-archiver/common/storage"
+	"github.com/base-org/blob-archiver/validator/flags"
 	"github.com/ethereum-optimism/optimism/op-service/retry"
 	"github.com/ethereum/go-ethereum/log"
 )
@@ -29,14 +30,21 @@ const (
 	retryAttempts = 10
 )
 
-func NewValidator(l log.Logger, headerClient client.BeaconBlockHeadersProvider, beaconAPI BlobSidecarClient, blobAPI BlobSidecarClient, app context.CancelCauseFunc, numBlocks int) *ValidatorService {
+var (
+	formatSettingToHeader = map[string]Format{
+		"json": FormatJson,
+		"ssz":  FormatSSZ,
+	}
+)
+
+func NewValidator(l log.Logger, headerClient client.BeaconBlockHeadersProvider, beaconAPI BlobSidecarClient, blobAPI BlobSidecarClient, app context.CancelCauseFunc, cfg flags.ValidatorConfig) *ValidatorService {
 	return &ValidatorService{
 		log:          l,
 		headerClient: headerClient,
 		beaconAPI:    beaconAPI,
 		blobAPI:      blobAPI,
 		closeApp:     app,
-		numBlocks:    numBlocks,
+		cfg:          cfg,
 	}
 }
 
@@ -47,7 +55,7 @@ type ValidatorService struct {
 	beaconAPI    BlobSidecarClient
 	blobAPI      BlobSidecarClient
 	closeApp     context.CancelCauseFunc
-	numBlocks    int
+	cfg          flags.ValidatorConfig
 }
 
 // Start starts the validator service. This will fetch the current range of blocks to validate and start the validation
@@ -64,7 +72,7 @@ func (a *ValidatorService) Start(ctx context.Context) error {
 	}
 
 	end := header.Data.Header.Message.Slot - finalizedL1Offset
-	start := end - phase0.Slot(a.numBlocks)
+	start := end - phase0.Slot(a.cfg.NumBlocks)
 
 	go a.checkBlobs(ctx, start, end)
 
@@ -126,7 +134,9 @@ func (a *ValidatorService) checkBlobs(ctx context.Context, start phase0.Slot, en
 	var result CheckBlobResult
 
 	for slot := start; slot <= end; slot++ {
-		for _, format := range []Format{FormatJson, FormatSSZ} {
+		for _, setting := range a.cfg.ValidateFormats {
+			format := formatSettingToHeader[setting]
+
 			id := strconv.FormatUint(uint64(slot), 10)
 
 			l := a.log.New("format", format, "slot", slot)

validator/service/service_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/base-org/blob-archiver/common/beacon/beacontest"
 	"github.com/base-org/blob-archiver/common/blobtest"
 	"github.com/base-org/blob-archiver/common/storage"
+	"github.com/base-org/blob-archiver/validator/flags"
 	"github.com/ethereum-optimism/optimism/op-service/testlog"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/stretchr/testify/require"
@@ -71,9 +72,10 @@ func setup(t *testing.T) (*ValidatorService, *beacontest.StubBeaconClient, *stub
 		data: make(map[string]response),
 	}
 
-	numBlocks := 600
-
-	return NewValidator(l, headerClient, beacon, blob, cancel, numBlocks), headerClient, beacon, blob
+	return NewValidator(l, headerClient, beacon, blob, cancel, flags.ValidatorConfig{
+		ValidateFormats: []string{"json", "ssz"},
+		NumBlocks:       600,
+	}), headerClient, beacon, blob
 }
 
 func TestValidatorService_OnFetchError(t *testing.T) {