feature: add caching-proxy clean subcommand

CHANGELOG.md

@@ -7,6 +7,7 @@ project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ### Added
 
+* add `caching-proxy clean <dsn> "error:<regex>"` subcommand to remove wrongfully-cached error entries whose message matches the regex, with `--batch-size` (1 deletes serially) and `--dry-run` flags; prints a styled summary report (auto-plain when not a TTY)
 * add fallback endpoint support with `--fallback` flag, with `--request-timeout` and `--timeout-before-fallback`
 
 ### Changed

cmd/evmx/caching_proxy.go

@@ -31,7 +31,7 @@ import (
 	"go.uber.org/zap"
 )
 
-var CachingProxyCmd = cli.Command(cachingProxyE,
+var CachingProxyCmd = cli.Group(
 	"caching-proxy",
 	"Starts the JSON-RPC caching proxy",
 	cli.Description(`
@@ -52,7 +52,9 @@ var CachingProxyCmd = cli.Command(cachingProxyE,
 		flags.Duration("shutdown-signal-delay", 0, "Upon shutdown signal, the server will return errors to the healthchecks endpoint, but still serve traffic for that time duration")
 		flags.StringSliceP("headers", "H", []string{}, "Headers to add to upstream eth_call requests (format: 'Header-Name: value')")
 	}),
+	cli.Execute(cachingProxyE),
 	cli.ExactArgs(2),
+	CachingProxyCleanCmd,
 )
 
 func cachingProxyE(cmd *cobra.Command, args []string) error {

cmd/evmx/caching_proxy_clean.go

@@ -0,0 +1,124 @@
+package main
+
+import (
+	"fmt"
+	"regexp"
+	"strings"
+
+	"github.com/spf13/cobra"
+	"github.com/spf13/pflag"
+	"github.com/streamingfast/cli"
+	"github.com/streamingfast/cli/sflags"
+	"github.com/streamingfast/evmx/cmd/evmx/stylex"
+	"github.com/streamingfast/evmx/executor"
+	"github.com/streamingfast/kvdb/store"
+	"go.uber.org/zap"
+)
+
+const errorMatcherPrefix = "error:"
+
+var CachingProxyCleanCmd = cli.Command(cachingProxyCleanE,
+	"clean",
+	"Removes wrongfully-cached entries from the caching proxy store",
+	cli.Description(`
+		Scans every entry in the caching proxy KV store and removes cached error entries whose
+		message matches the provided matcher.
+
+		The matcher must be of the form 'error:<regex>' where <regex> is a Go regular expression
+		tested against the cached error message. The 'error:' prefix is reserved so that other
+		matcher kinds can be added later.
+
+		ex: 'evmx caching-proxy clean bigkv://my-proj.my-instance/my-table "error:execution reverted"'
+	`),
+	cli.Flags(func(flags *pflag.FlagSet) {
+		flags.Int("batch-size", 100, "Number of keys to delete per batch, a value of 1 deletes serially (one key per call), bypassing batching")
+		flags.Bool("dry-run", false, "When true, scans and reports matching entries but deletes nothing")
+	}),
+	cli.ExactArgs(2),
+)
+
+func cachingProxyCleanE(cmd *cobra.Command, args []string) error {
+	dsn := args[0]
+	matcher := args[1]
+
+	regex, err := parseErrorMatcher(matcher)
+	if err != nil {
+		return err
+	}
+
+	batchSize := sflags.MustGetInt(cmd, "batch-size")
+	if batchSize < 1 {
+		return fmt.Errorf("batch-size must be at least 1, got %d", batchSize)
+	}
+	dryRun := sflags.MustGetBool(cmd, "dry-run")
+
+	kv, err := store.New(dsn)
+	if err != nil {
+		return fmt.Errorf("open store %q: %w", dsn, err)
+	}
+	defer kv.Close()
+
+	zlog.Info("cleaning caching proxy error entries",
+		zap.String("dsn", dsn),
+		zap.String("message_regex", regex.String()),
+		zap.Int("batch_size", batchSize),
+		zap.Bool("dry_run", dryRun),
+	)
+
+	summary, err := executor.CleanErrors(cmd.Context(), kv, regex, batchSize, dryRun, zlog)
+	if err != nil {
+		return fmt.Errorf("clean errors: %w", err)
+	}
+
+	fmt.Println(renderCleanReport(summary, dryRun))
+
+	return nil
+}
+
+// renderCleanReport formats a [executor.CleanErrorsSummary] into a styled, human-friendly
+// report. Styling is automatically disabled when standard output is not a TTY.
+func renderCleanReport(summary *executor.CleanErrorsSummary, dryRun bool) string {
+	var b strings.Builder
+
+	title := "Caching Proxy Clean Report"
+	b.WriteString(stylex.Title(title))
+	b.WriteByte('\n')
+	b.WriteString(stylex.Dim(strings.Repeat("─", len(title))))
+	b.WriteByte('\n')
+
+	row := func(label string, value string) {
+		fmt.Fprintf(&b, "  %s  %s\n", stylex.Label(fmt.Sprintf("%-14s", label)), value)
+	}
+
+	row("Scanned", stylex.Valuef("%d", summary.Scanned))
+	row("Error entries", stylex.Valuef("%d", summary.Errors))
+	row("Matched", stylex.Valuef("%d", summary.Matched))
+
+	deleted := stylex.Successf("%d", summary.Deleted)
+	if dryRun {
+		deleted = stylex.Warnf("%d", summary.Deleted) + " " + stylex.Note("(dry-run, nothing deleted)")
+	}
+	row("Deleted", deleted)
+
+	return b.String()
+}
+
+// parseErrorMatcher parses a matcher of the form 'error:<regex>' into a compiled regular
+// expression tested against cached error messages.
+func parseErrorMatcher(matcher string) (*regexp.Regexp, error) {
+	if !strings.HasPrefix(matcher, errorMatcherPrefix) {
+		return nil, fmt.Errorf("invalid matcher %q, expected form 'error:<regex>'", matcher)
+	}
+
+	pattern := strings.TrimPrefix(matcher, errorMatcherPrefix)
+	if pattern == "" {
+		return nil, fmt.Errorf("invalid matcher %q, the regex after 'error:' must not be empty", matcher)
+	}
+
+	regex, err := regexp.Compile(pattern)
+	if err != nil {
+		return nil, fmt.Errorf("invalid regex %q: %w", pattern, err)
+	}
+
+	return regex, nil
+}

cmd/evmx/caching_proxy_clean_test.go

@@ -0,0 +1,54 @@
+package main
+
+import (
+	"testing"
+
+	"github.com/streamingfast/evmx/executor"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRenderCleanReport(t *testing.T) {
+	dryRun := renderCleanReport(&executor.CleanErrorsSummary{Scanned: 9508, Errors: 7736, Matched: 7727, Deleted: 0}, true)
+	assert.Contains(t, dryRun, "9508")
+	assert.Contains(t, dryRun, "7736")
+	assert.Contains(t, dryRun, "7727")
+	assert.Contains(t, dryRun, "Scanned")
+	assert.Contains(t, dryRun, "Matched")
+	assert.Contains(t, dryRun, "Deleted")
+	assert.Contains(t, dryRun, "dry-run", "dry-run report must flag that nothing was deleted")
+
+	real := renderCleanReport(&executor.CleanErrorsSummary{Scanned: 10, Errors: 4, Matched: 3, Deleted: 3}, false)
+	assert.Contains(t, real, "Deleted")
+	assert.NotContains(t, real, "dry-run")
+}
+
+func TestParseErrorMatcher(t *testing.T) {
+	tests := []struct {
+		name       string
+		matcher    string
+		expectErr  bool
+		matchInput string
+		matches    bool
+	}{
+		{"valid regex", "error:execution reverted", false, "execution reverted: boom", true},
+		{"valid regex no match", "error:execution reverted", false, "insufficient funds", false},
+		{"missing prefix", "execution reverted", true, "", false},
+		{"wrong prefix", "data:foo", true, "", false},
+		{"empty regex", "error:", true, "", false},
+		{"invalid regex", "error:[unterminated", true, "", false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			regex, err := parseErrorMatcher(tt.matcher)
+			if tt.expectErr {
+				require.Error(t, err)
+				return
+			}
+
+			require.NoError(t, err)
+			assert.Equal(t, tt.matches, regex.MatchString(tt.matchInput))
+		})
+	}
+}

cmd/evmx/stylex/doc.go

@@ -0,0 +1,9 @@
+// Package stylex provides lipgloss styles for the evmx CLI commands.
+//
+// It defines a small set of lipgloss styles that can be used across commands to
+// ensure a consistent look and feel. Styling is automatically disabled when the
+// output is not a TTY, so commands can be used in scripts without worrying about
+// ANSI escape codes.
+//
+// You have nothing to do to activate that.
+package stylex

cmd/evmx/stylex/stylex.go

@@ -0,0 +1,65 @@
+package stylex
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/charmbracelet/lipgloss"
+	"github.com/mattn/go-isatty"
+)
+
+var (
+	// IsTTY is true when standard output is a terminal, in which case styling is enabled.
+	IsTTY = isatty.IsTerminal(os.Stdout.Fd())
+
+	TitleStyle  = lipgloss.NewStyle().Bold(true).Foreground(lipgloss.Color("12"))
+	HeaderStyle = lipgloss.NewStyle().Bold(true).Foreground(lipgloss.Color("14"))
+	NoteStyle   = lipgloss.NewStyle().Foreground(lipgloss.Color("238")).Italic(true)
+	DimStyle    = lipgloss.NewStyle().Foreground(lipgloss.Color("240"))
+
+	LabelStyle = lipgloss.NewStyle().Foreground(lipgloss.Color("245"))
+	ValueStyle = lipgloss.NewStyle().Bold(true)
+
+	SuccessStyle = lipgloss.NewStyle().Foreground(lipgloss.Color("10"))  // Green
+	WarnStyle    = lipgloss.NewStyle().Foreground(lipgloss.Color("208")) // Orange
+	ErrorStyle   = lipgloss.NewStyle().Foreground(lipgloss.Color("9"))   // Red
+)
+
+func init() {
+	// Disable styling if not a TTY
+	if !IsTTY {
+		DisableColors()
+	}
+}
+
+// DisableColors resets every style to a no-op style, producing plain unstyled output.
+func DisableColors() {
+	TitleStyle = lipgloss.NewStyle()
+	HeaderStyle = lipgloss.NewStyle()
+	NoteStyle = lipgloss.NewStyle()
+	DimStyle = lipgloss.NewStyle()
+	LabelStyle = lipgloss.NewStyle()
+	ValueStyle = lipgloss.NewStyle()
+	SuccessStyle = lipgloss.NewStyle()
+	WarnStyle = lipgloss.NewStyle()
+	ErrorStyle = lipgloss.NewStyle()
+}
+
+func Titlef(format string, a ...any) string   { return TitleStyle.Render(fmt.Sprintf(format, a...)) }
+func Title(str string) string                 { return TitleStyle.Render(str) }
+func Headerf(format string, a ...any) string  { return HeaderStyle.Render(fmt.Sprintf(format, a...)) }
+func Header(str string) string                { return HeaderStyle.Render(str) }
+func Notef(format string, a ...any) string    { return NoteStyle.Render(fmt.Sprintf(format, a...)) }
+func Note(str string) string                  { return NoteStyle.Render(str) }
+func Dimf(format string, a ...any) string     { return DimStyle.Render(fmt.Sprintf(format, a...)) }
+func Dim(str string) string                   { return DimStyle.Render(str) }
+func Labelf(format string, a ...any) string   { return LabelStyle.Render(fmt.Sprintf(format, a...)) }
+func Label(str string) string                 { return LabelStyle.Render(str) }
+func Valuef(format string, a ...any) string   { return ValueStyle.Render(fmt.Sprintf(format, a...)) }
+func Value(str string) string                 { return ValueStyle.Render(str) }
+func Successf(format string, a ...any) string { return SuccessStyle.Render(fmt.Sprintf(format, a...)) }
+func Success(str string) string               { return SuccessStyle.Render(str) }
+func Warnf(format string, a ...any) string    { return WarnStyle.Render(fmt.Sprintf(format, a...)) }
+func Warn(str string) string                  { return WarnStyle.Render(str) }
+func Errorf(format string, a ...any) string   { return ErrorStyle.Render(fmt.Sprintf(format, a...)) }
+func Error(str string) string                 { return ErrorStyle.Render(str) }

executor/caching_proxy_clean.go

@@ -0,0 +1,107 @@
+package executor
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"regexp"
+
+	"github.com/streamingfast/eth-go/rpc"
+	"github.com/streamingfast/kvdb/store"
+	"go.uber.org/zap"
+)
+
+// CleanErrorsSummary reports the outcome of a [CleanErrors] run.
+type CleanErrorsSummary struct {
+	Scanned int // Total number of entries scanned in the store
+	Errors  int // Number of error entries seen (value prefixed with ErrorPrefix)
+	Matched int // Number of error entries whose message matched the regex
+	Deleted int // Number of entries actually deleted (always 0 when dryRun)
+}
+
+// CleanErrors scans every entry in the store and deletes error entries (those whose
+// value is prefixed with [ErrorPrefix]) whose rpc error message matches messageRegex.
+//
+// Deletions are batched in groups of batchSize; a batchSize of 1 deletes serially,
+// one key per [store.KVStore.BatchDelete] call, bypassing batching. When dryRun is
+// true, matches are counted but nothing is deleted.
+func CleanErrors(
+	ctx context.Context,
+	kv store.KVStore,
+	messageRegex *regexp.Regexp,
+	batchSize int,
+	dryRun bool,
+	logger *zap.Logger,
+) (*CleanErrorsSummary, error) {
+	if batchSize < 1 {
+		return nil, fmt.Errorf("batch size must be at least 1, got %d", batchSize)
+	}
+
+	summary := &CleanErrorsSummary{}
+	batch := make([][]byte, 0, batchSize)
+
+	flush := func() error {
+		if len(batch) == 0 {
+			return nil
+		}
+		if err := kv.BatchDelete(ctx, batch); err != nil {
+			return fmt.Errorf("batch delete: %w", err)
+		}
+		summary.Deleted += len(batch)
+		batch = batch[:0]
+		return nil
+	}
+
+	// A nil prefix matches every key, giving us a full scan regardless of key length.
+	itr := kv.Prefix(ctx, nil, store.Unlimited)
+	for itr.Next() {
+		item := itr.Item()
+		summary.Scanned++
+
+		value := item.Value
+		if len(value) == 0 || value[0] != ErrorPrefix {
+			continue
+		}
+		summary.Errors++
+
+		rpcErr := &rpc.ErrResponse{}
+		if err := json.Unmarshal(value[1:], rpcErr); err != nil {
+			logger.Debug("skipping error entry with unparseable payload", zap.Error(err))
+			continue
+		}
+
+		if !messageRegex.MatchString(rpcErr.Message) {
+			continue
+		}
+		summary.Matched++
+
+		logger.Debug("matched error entry to clean",
+			zap.Int("code", int(rpcErr.Code)),
+			zap.String("message", rpcErr.Message),
+		)
+
+		if dryRun {
+			continue
+		}
+
+		// The iterator may reuse the key slice, copy it before retaining it in the batch.
+		key := make([]byte, len(item.Key))
+		copy(key, item.Key)
+		batch = append(batch, key)
+
+		if len(batch) >= batchSize {
+			if err := flush(); err != nil {
+				return summary, err
+			}
+		}
+	}
+	if err := itr.Err(); err != nil {
+		return summary, fmt.Errorf("scanning store: %w", err)
+	}
+
+	if err := flush(); err != nil {
+		return summary, err
+	}
+
+	return summary, nil
+}