Remove observable VMs from Geyser worker

Author: jeffyanta

CLAUDE.md

@@ -16,7 +16,7 @@ go test -run TestName ./path/to/package/...
 
 # Run locally (Docker)
 DATA_STORAGE_TYPE=memory make run-rpc
-DATA_STORAGE_TYPE=memory GEYSER_WORKER_GRPC_PLUGIN_ENDPOINT=localhost:10000 GEYSER_WORKER_VM_ACCOUNT=<VM_KEY> SOLANA_RPC_ENDPOINT=http://localhost:8899 make run-geyser
+DATA_STORAGE_TYPE=memory GEYSER_WORKER_GRPC_PLUGIN_ENDPOINT=localhost:10000 SOLANA_RPC_ENDPOINT=http://localhost:8899 make run-geyser
 ```
 
 Proto generation (`make generate`) is currently broken due to optional fields in the Geyser proto.

Makefile

@@ -67,7 +67,6 @@ run-geyser: build-geyser build-geyser-image
 		-e DATA_STORAGE_TYPE=$(DATA_STORAGE_TYPE) \
 		-e GEYSER_WORKER_GRPC_PLUGIN_ENDPOINT=$(GEYSER_WORKER_GRPC_PLUGIN_ENDPOINT) \
 		-e GEYSER_WORKER_PROGRAM_UPDATE_WORKER_COUNT=4 \
-		-e GEYSER_WORKER_VM_ACCOUNT=$(GEYSER_WORKER_VM_ACCOUNT) \
 		-e POSTGRES_USER=$(POSTGRES_USER) \
 		-e POSTGRES_PASSWORD=$(POSTGRES_PASSWORD) \
 		-e POSTGRES_HOST=$(POSTGRES_HOST) \

README.md

@@ -42,7 +42,7 @@ DATA_STORAGE_TYPE=memory make run-rpc
 5. Run the Geyser worker locally:
 
 ```bash
-DATA_STORAGE_TYPE=memory GEYSER_WORKER_GRPC_PLUGIN_ENDPOINT=localhost:10000 GEYSER_WORKER_VM_ACCOUNT=<VM_PUBLIC_KEY> SOLANA_RPC_ENDPOINT=http://localhost:8899 make run-geyser
+DATA_STORAGE_TYPE=memory GEYSER_WORKER_GRPC_PLUGIN_ENDPOINT=localhost:10000 SOLANA_RPC_ENDPOINT=http://localhost:8899 make run-geyser
 ```
 
 ## Getting Help

data/ram/memory/memory.go

@@ -57,17 +57,16 @@ func (s *store) Save(_ context.Context, data *ram.Record) error {
 }
 
 // GetAllMemoryAccounts implements ram.Store.GetAllMemoryAccounts
-func (s *store) GetAllMemoryAccounts(ctx context.Context, vm string) ([]string, error) {
+func (s *store) GetAllMemoryAccounts(ctx context.Context) ([]string, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	items := s.findByVm(vm)
-	if len(items) == 0 {
+	if len(s.records) == 0 {
 		return nil, ram.ErrAccountNotFound
 	}
 
 	uniqueMemoryAccounts := make(map[string]any)
-	for _, item := range items {
+	for _, item := range s.records {
 		uniqueMemoryAccounts[item.MemoryAccount] = struct{}{}
 	}
 
@@ -126,16 +125,6 @@ func (s *store) findByMemoryAccount(memoryAccount string) []*ram.Record {
 	return res
 }
 
-func (s *store) findByVm(vm string) []*ram.Record {
-	var res []*ram.Record
-	for _, item := range s.records {
-		if item.Vm == vm {
-			res = append(res, item)
-		}
-	}
-	return res
-}
-
 func (s *store) findByVmAddressAndAccountType(vm, address string, accountType vm.VirtualAccountType) []*ram.Record {
 	var res []*ram.Record
 	for _, item := range s.records {

data/ram/postgres/model.go

@@ -133,17 +133,15 @@ func (m *model) dbSave(ctx context.Context, tableName string, db *sqlx.DB) error
 	return pgutil.CheckNoRows(err, ram.ErrStaleState)
 }
 
-func dbGetAllMemoryAccounts(ctx context.Context, tableName string, db *sqlx.DB, vm string) ([]string, error) {
+func dbGetAllMemoryAccounts(ctx context.Context, tableName string, db *sqlx.DB) ([]string, error) {
 	res := []string{}
 
-	query := `SELECT DISTINCT(memory_account) FROM ` + tableName + `
-		WHERE vm = $1`
+	query := `SELECT DISTINCT(memory_account) FROM ` + tableName
 
 	err := db.SelectContext(
 		ctx,
 		&res,
 		query,
-		vm,
 	)
 	if err != nil {
 		return nil, pgutil.CheckNoRows(err, ram.ErrAccountNotFound)

data/ram/postgres/store.go

@@ -41,8 +41,8 @@ func (s *store) Save(ctx context.Context, record *ram.Record) error {
 }
 
 // GetAllMemoryAccounts implements ram.Store.GetAllMemoryAccounts
-func (s *store) GetAllMemoryAccounts(ctx context.Context, vm string) ([]string, error) {
-	return dbGetAllMemoryAccounts(ctx, s.tableName, s.db, vm)
+func (s *store) GetAllMemoryAccounts(ctx context.Context) ([]string, error) {
+	return dbGetAllMemoryAccounts(ctx, s.tableName, s.db)
 }
 
 // GetAllByMemoryAccount implements ram.Store.GetAllByMemoryAccount

data/ram/store.go

@@ -18,7 +18,7 @@ type Store interface {
 	Save(ctx context.Context, record *Record) error
 
 	// GetAllMemoryAccounts gets all unique memory account addresses
-	GetAllMemoryAccounts(ctx context.Context, vm string) ([]string, error)
+	GetAllMemoryAccounts(ctx context.Context) ([]string, error)
 
 	// GetAllByMemoryAccount gets all database records for a given memory account
 	GetAllByMemoryAccount(ctx context.Context, memoryAccount string) ([]*Record, error)

data/ram/tests/tests.go

@@ -106,7 +106,7 @@ func testGetAllMemoryAccounts(t *testing.T, s ram.Store) {
 	t.Run("testGetAllMemoryAccounts", func(t *testing.T) {
 		ctx := context.Background()
 
-		_, err := s.GetAllMemoryAccounts(ctx, "vm2")
+		_, err := s.GetAllMemoryAccounts(ctx)
 		assert.Equal(t, ram.ErrAccountNotFound, err)
 
 		for i := 0; i < 3; i++ {
@@ -121,10 +121,9 @@ func testGetAllMemoryAccounts(t *testing.T, s ram.Store) {
 			}))
 		}
 
-		actual, err := s.GetAllMemoryAccounts(ctx, "vm2")
+		actual, err := s.GetAllMemoryAccounts(ctx)
 		require.NoError(t, err)
-		require.Len(t, actual, 1)
-		assert.Equal(t, "memory_account_2", actual[0])
+		require.Len(t, actual, 3)
 	})
 }
 

geyser/config.go

@@ -1,8 +1,6 @@
 package geyser
 
 import (
-	"context"
-	"strings"
 	"time"
 
 	"github.com/code-payments/ocp-server/config"
@@ -24,9 +22,6 @@ const (
 	ProgramUpdateQueueSizeConfigEnvName = envConfigPrefix + "PROGRAM_UPDATE_QUEUE_SIZE"
 	defaultProgramUpdateQueueSize       = 1_000_000
 
-	VmAccountsConfigEnvName = envConfigPrefix + "VM_ACCOUNTS"
-	defaultVmAccounts       = ""
-
 	MemoryAccountBackupWorkerIntervalConfigEnvName = envConfigPrefix + "MEMORY_ACCOUNT_BACKUP_WORKER_INTERVAL"
 	defaultMemoryAccountBackupWorkerInterval       = time.Minute
 )
@@ -38,8 +33,6 @@ type conf struct {
 	programUpdateWorkerCount config.Uint64
 	programUpdateQueueSize   config.Uint64
 
-	vmAccounts config.String
-
 	memoryAccountBackkupWorkerInterval config.Duration
 }
 
@@ -56,17 +49,8 @@ func WithEnvConfigs() ConfigProvider {
 			programUpdateWorkerCount: env.NewUint64Config(ProgramUpdateWorkerCountConfigEnvName, defaultProgramUpdateWorkerCount),
 			programUpdateQueueSize:   env.NewUint64Config(ProgramUpdateQueueSizeConfigEnvName, defaultProgramUpdateQueueSize),
 
-			vmAccounts: env.NewStringConfig(VmAccountsConfigEnvName, defaultVmAccounts),
-
 			memoryAccountBackkupWorkerInterval: env.NewDurationConfig(MemoryAccountBackupWorkerIntervalConfigEnvName, defaultMemoryAccountBackupWorkerInterval),
 		}
 	}
 }
 
-func parseVmAccountsConfig(ctx context.Context, c *conf) []string {
-	parts := strings.Split(c.vmAccounts.Get(ctx), ",")
-	for i, part := range parts {
-		parts[i] = strings.TrimSpace(part)
-	}
-	return parts
-}

geyser/handler.go

@@ -34,7 +34,6 @@ func initializeProgramAccountUpdateHandlers(log *zap.Logger, conf *conf, solanaC
 			solanaClient,
 			ramStore,
 			conf.memoryAccountBackkupWorkerInterval.Get(ctx),
-			parseVmAccountsConfig(ctx, conf)...,
 		),
 	}
 }