Merge pull request #72 from ConstellationCrypto/inomurko/fix-read-only-fallback

fix: enable read only

Author: tuxcanfly

celestia_server.go

@@ -99,7 +99,8 @@ type CelestiaServer struct {
 	metricsServer  *http.Server // Separate server for graceful shutdown
 
 	// Fallback provider for write-through and read-fallback
-	fallback fallback.Provider
+	fallback       fallback.Provider
+	fallbackMode   string
 
 	// Track pending async operations for graceful shutdown
 	wg sync.WaitGroup
@@ -119,6 +120,7 @@ func NewCelestiaServer(
 	metricsEnabled bool,
 	metricsPort int,
 	fallbackProvider fallback.Provider,
+	fallbackMode string,
 	log log.Logger,
 ) *CelestiaServer {
 	endpoint := net.JoinHostPort(host, strconv.Itoa(port))
@@ -138,6 +140,7 @@ func NewCelestiaServer(
 		metricsEnabled: metricsEnabled,
 		metricsPort:    metricsPort,
 		fallback:       fallbackProvider,
+		fallbackMode:   fallback.NormalizeMode(fallbackMode),
 		httpServer: &http.Server{
 			Addr:           endpoint,
 			ReadTimeout:    httpReadTimeout,
@@ -278,8 +281,8 @@ func (d *CelestiaServer) getBlob(ctx context.Context, comm []byte) ([]byte, erro
 
 	data, celestiaErr := d.store.Get(getCtx, comm)
 	if celestiaErr == nil {
-		// Success from Celestia - read-through to fallback for future requests
-		if d.fallback.Available() {
+		// Success from Celestia - optionally read-through to fallback for future requests
+		if d.fallback.Available() && fallback.WriteEnabled(d.fallbackMode) {
 			d.wg.Add(1)
 			go d.putFallback(context.Background(), comm, data)
 		}
@@ -443,7 +446,7 @@ func (d *CelestiaServer) HandlePut(w http.ResponseWriter, r *http.Request) {
 		"duration", duration)
 
 	// Write to fallback provider asynchronously (non-blocking)
-	if d.fallback.Available() {
+	if d.fallback.Available() && fallback.WriteEnabled(d.fallbackMode) {
 		d.wg.Add(1)
 		go d.putFallback(context.Background(), commitment, input)
 	}

celestia_server_test.go

@@ -103,11 +103,16 @@ func createTestServer(t *testing.T, store Store) *CelestiaServer {
 		false,           // metrics disabled for unit tests
 		0,
 		nil, // fallback provider (nil = NoopProvider)
+		fallback.ModeBoth,
 		logger,
 	)
 }
 
 func createTestServerWithFallback(t *testing.T, store Store, provider fallback.Provider) *CelestiaServer {
+	return createTestServerWithFallbackMode(t, store, provider, fallback.ModeBoth)
+}
+
+func createTestServerWithFallbackMode(t *testing.T, store Store, provider fallback.Provider, mode string) *CelestiaServer {
 	logger := log.New()
 
 	return NewCelestiaServer(
@@ -123,6 +128,7 @@ func createTestServerWithFallback(t *testing.T, store Store, provider fallback.P
 		false,
 		0,
 		provider,
+		mode,
 		logger,
 	)
 }
@@ -338,6 +344,23 @@ func TestGetBlob_DoesNotUseFallbackOnCelestiaNotFound(t *testing.T) {
 	assert.Equal(t, 0, fallbackProvider.getCalls)
 }
 
+func TestGetBlob_ReadFallbackMode_SkipsWriteThrough(t *testing.T) {
+	blob := []byte("blob from celestia")
+	fallbackProvider := &mockFallbackProvider{available: true}
+	store := &mockStore{
+		getFunc: func(ctx context.Context, key []byte) ([]byte, error) {
+			return blob, nil
+		},
+	}
+
+	server := createTestServerWithFallbackMode(t, store, fallbackProvider, fallback.ModeReadFallback)
+
+	data, err := server.getBlob(context.Background(), commitmentForData(t, blob))
+	require.NoError(t, err)
+	assert.Equal(t, blob, data)
+	assert.Equal(t, 0, fallbackProvider.putCalls)
+}
+
 func TestGetBlob_VerifiesFallbackDataAgainstCommitment(t *testing.T) {
 	commitment := commitmentForData(t, []byte("expected blob"))
 	store := &mockStore{

cmd/daserver/config.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	celestia "github.com/celestiaorg/op-alt-da"
+	"github.com/celestiaorg/op-alt-da/fallback"
 	"github.com/celestiaorg/op-alt-da/signer"
 )
 
@@ -119,6 +120,9 @@ type FallbackConfig struct {
 	// Provider type: "s3"
 	Provider string `toml:"provider"`
 
+	// Mode controls fallback write/read behavior: write_through, read_fallback, or both.
+	Mode string `toml:"mode"`
+
 	// S3 configuration (when provider = "s3")
 	S3 S3Config `toml:"s3"`
 }
@@ -183,6 +187,7 @@ func DefaultConfig() Config {
 		Fallback: FallbackConfig{
 			Enabled:  false,
 			Provider: "s3",
+			Mode:     fallback.ModeBoth,
 			S3: S3Config{
 				Region:  "us-east-1",
 				Timeout: "30s",
@@ -252,6 +257,10 @@ func (c *Config) Validate() error {
 		}
 	}
 
+	if err := fallback.ValidateMode(c.Fallback.Mode); err != nil {
+		return err
+	}
+
 	return nil
 }
 

cmd/daserver/config_builder.go

@@ -150,6 +150,9 @@ func BuildConfigFromCLI(ctx *cli.Context) (*Config, error) {
 	if ctx.IsSet(FallbackProviderFlagName) {
 		cfg.Fallback.Provider = ctx.String(FallbackProviderFlagName)
 	}
+	if ctx.IsSet(FallbackModeFlagName) {
+		cfg.Fallback.Mode = ctx.String(FallbackModeFlagName)
+	}
 	if ctx.IsSet(FallbackS3BucketFlagName) {
 		cfg.Fallback.S3.Bucket = ctx.String(FallbackS3BucketFlagName)
 	}

cmd/daserver/entrypoint.go

@@ -121,6 +121,7 @@ func StartDAServer(cliCtx *cli.Context) error {
 
 		// Initialize fallback provider (prefer TOML config, fall back to CLI flags)
 		var fallbackProvider fallback.Provider = &fallback.NoopProvider{}
+		fallbackMode := firstNonEmpty(cfg.Fallback.Mode, cliCtx.String(FallbackModeFlagName))
 		fallbackEnabled := cfg.Fallback.Enabled || cliCtx.Bool(FallbackEnabledFlagName)
 		if fallbackEnabled {
 			provider := cfg.Fallback.Provider
@@ -162,10 +163,12 @@ func StartDAServer(cliCtx *cli.Context) error {
 					return fmt.Errorf("failed to initialize S3 fallback provider: %w", err)
 				}
 				fallbackProvider = s3Provider
+				fallbackMode := firstNonEmpty(cfg.Fallback.Mode, cliCtx.String(FallbackModeFlagName))
 				l.Info("Fallback provider initialized",
 					"provider", "s3",
 					"bucket", s3Cfg.Bucket,
 					"prefix", s3Cfg.Prefix,
+					"mode", fallback.NormalizeMode(fallbackMode),
 					"read_legacy_blobs", s3Cfg.ReadLegacyBlobs)
 			default:
 				return fmt.Errorf("unknown fallback provider: %s", provider)
@@ -185,6 +188,7 @@ func StartDAServer(cliCtx *cli.Context) error {
 			cfg.Metrics.Enabled,
 			cfg.Metrics.Port,
 			fallbackProvider,
+			fallbackMode,
 			l,
 		)
 	} else {

cmd/daserver/flags.go

@@ -10,6 +10,7 @@ import (
 	"github.com/urfave/cli/v2"
 
 	celestia "github.com/celestiaorg/op-alt-da"
+	"github.com/celestiaorg/op-alt-da/fallback"
 	"github.com/celestiaorg/op-alt-da/fallback/s3"
 	"github.com/celestiaorg/op-alt-da/signer"
 	opservice "github.com/ethereum-optimism/optimism/op-service"
@@ -47,6 +48,7 @@ const (
 	// fallback provider flags
 	FallbackEnabledFlagName     = "fallback.enabled"
 	FallbackProviderFlagName    = "fallback.provider"
+	FallbackModeFlagName        = "fallback.mode"
 	FallbackS3BucketFlagName    = "fallback.s3.bucket"
 	FallbackS3PrefixFlagName    = "fallback.s3.prefix"
 	FallbackS3EndpointFlagName  = "fallback.s3.endpoint"
@@ -193,6 +195,12 @@ var (
 		Value:   "s3",
 		EnvVars: prefixEnvVars("FALLBACK_PROVIDER"),
 	}
+	FallbackModeFlag = &cli.StringFlag{
+		Name:    FallbackModeFlagName,
+		Usage:   "Fallback mode: write_through, read_fallback, or both",
+		Value:   fallback.ModeBoth,
+		EnvVars: prefixEnvVars("FALLBACK_MODE"),
+	}
 	FallbackS3BucketFlag = &cli.StringFlag{
 		Name:    FallbackS3BucketFlagName,
 		Usage:   "S3 bucket name for fallback storage",
@@ -277,6 +285,7 @@ var optionalFlags = []cli.Flag{
 	// Fallback flags
 	FallbackEnabledFlag,
 	FallbackProviderFlag,
+	FallbackModeFlag,
 	FallbackS3BucketFlag,
 	FallbackS3PrefixFlag,
 	FallbackS3EndpointFlag,
@@ -300,6 +309,7 @@ func init() {
 type CLIFallbackConfig struct {
 	Enabled  bool
 	Provider string
+	Mode     string
 	S3       s3.Config
 }
 
@@ -364,6 +374,7 @@ func ReadCLIConfig(ctx *cli.Context) CLIConfig {
 		Fallback: CLIFallbackConfig{
 			Enabled:  ctx.Bool(FallbackEnabledFlagName),
 			Provider: ctx.String(FallbackProviderFlagName),
+			Mode:     ctx.String(FallbackModeFlagName),
 			S3: s3.Config{
 				Bucket:          ctx.String(FallbackS3BucketFlagName),
 				Prefix:          ctx.String(FallbackS3PrefixFlagName),

fallback/mode.go

@@ -0,0 +1,42 @@
+package fallback
+
+import "fmt"
+
+const (
+	ModeWriteThrough = "write_through"
+	ModeReadFallback = "read_fallback"
+	ModeBoth         = "both"
+)
+
+// NormalizeMode returns the effective fallback mode, defaulting to both.
+func NormalizeMode(mode string) string {
+	if mode == "" {
+		return ModeBoth
+	}
+	return mode
+}
+
+// WriteEnabled reports whether fallback writes (write-through) are enabled.
+func WriteEnabled(mode string) bool {
+	switch NormalizeMode(mode) {
+	case ModeReadFallback:
+		return false
+	case ModeWriteThrough, ModeBoth:
+		return true
+	default:
+		return true
+	}
+}
+
+// ValidateMode checks that mode is one of the supported values.
+func ValidateMode(mode string) error {
+	if mode == "" {
+		return nil
+	}
+	switch mode {
+	case ModeWriteThrough, ModeReadFallback, ModeBoth:
+		return nil
+	default:
+		return fmt.Errorf("fallback.mode must be %q, %q, or %q", ModeWriteThrough, ModeReadFallback, ModeBoth)
+	}
+}

fallback/mode_test.go

@@ -0,0 +1,25 @@
+package fallback
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestNormalizeMode(t *testing.T) {
+	require.Equal(t, ModeBoth, NormalizeMode(""))
+	require.Equal(t, ModeReadFallback, NormalizeMode(ModeReadFallback))
+}
+
+func TestWriteEnabled(t *testing.T) {
+	require.True(t, WriteEnabled(ModeBoth))
+	require.True(t, WriteEnabled(ModeWriteThrough))
+	require.False(t, WriteEnabled(ModeReadFallback))
+	require.True(t, WriteEnabled(""))
+}
+
+func TestValidateMode(t *testing.T) {
+	require.NoError(t, ValidateMode(""))
+	require.NoError(t, ValidateMode(ModeBoth))
+	require.Error(t, ValidateMode("invalid"))
+}