From 0d9bc44bec2da3150dadf6941383dc2850f4a6e7 Mon Sep 17 00:00:00 2001 From: Marcus Pasell <3690498+rickyrombo@users.noreply.github.com> Date: Fri, 23 Oct 2026 00:00:18 -0700 Subject: [PATCH] fix(api): stamp the new chain ID on flushed transactions go-openaudio#601 makes core refuse a ForwardTransaction whose chain ID differs from its own, and on audius-mainnet-beta also one with no chain ID at all. That closes the leak of the old chain's gossip into the new chain's mempool, but the flusher forwards the same way and stamps nothing, so after #601 the new chain would refuse every queued row. The flusher now stamps NewChainID on each forward. It defaults to audius-mainnet-beta in prod and openaudio-devnet in dev, and newChainId overrides it. It is configured rather than read back from newChainUrl so the check still means something: pointed at a node on another chain, the flusher stalls on the first refused row instead of writing the backlog there. With flushing enabled and no chain ID, the flusher logs and does not start. Bumps go-openaudio to #601's head for the chain_id field. Co-Authored-By: Claude Opus 5.5 --- api/new_chain_flusher.go | 4 ++++ api/new_chain_flusher_test.go | 7 ++++++- api/server.go | 6 +++++- config/config.go | 10 ++++++++++ go.mod | 2 +- go.sum | 4 ++-- 6 files changed, 28 insertions(+), 5 deletions(-) diff --git a/api/new_chain_flusher.go b/api/new_chain_flusher.go index 3fc38487..ddef5248 100644 --- a/api/new_chain_flusher.go +++ b/api/new_chain_flusher.go @@ -193,6 +193,9 @@ func (f *NewChainFlusher) flushRow(ctx context.Context, row queueRow) error { return nil } + // The chain ID tells core which chain this forward is meant for. A node on + // any other chain refuses it, so a newChainUrl pointed at the wrong chain + // stalls here instead of writing the backlog there. req := connect.NewRequest(&v1.ForwardTransactionRequest{ Transaction: &v1.SignedTransaction{ RequestId: uuid.NewString(), @@ -200,6 +203,7 @@ func (f *NewChainFlusher) flushRow(ctx context.Context, row queueRow) error { ManageEntity: &tx, }, }, + ChainId: f.cfg.NewChainID, }) const ( diff --git a/api/new_chain_flusher_test.go b/api/new_chain_flusher_test.go index aa3833c1..296dfd88 100644 --- a/api/new_chain_flusher_test.go +++ b/api/new_chain_flusher_test.go @@ -22,11 +22,13 @@ import ( type mockCoreService struct { corev1connect.UnimplementedCoreServiceHandler received []*corev1.ManageEntityLegacy + chainIDs []string calls atomic.Int32 } func (m *mockCoreService) ForwardTransaction(_ context.Context, req *connect.Request[corev1.ForwardTransactionRequest]) (*connect.Response[corev1.ForwardTransactionResponse], error) { m.calls.Add(1) + m.chainIDs = append(m.chainIDs, req.Msg.ChainId) if me := req.Msg.Transaction.GetManageEntity(); me != nil { m.received = append(m.received, me) } @@ -216,7 +218,7 @@ func TestNewChainFlusherTrim(t *testing.T) { // TestNewChainFlusherSends verifies that the flusher forwards all queued rows to // the new chain and marks them flushed — retaining them — on success. func TestNewChainFlusherSends(t *testing.T) { - cfg := &config.Config{} // no trim + cfg := &config.Config{NewChainID: "audius-mainnet-beta"} // no trim f, mock := newTestFlusher(t, cfg) block10 := int64(10) @@ -251,6 +253,9 @@ func TestNewChainFlusherSends(t *testing.T) { receivedIDs[i] = me.EntityId } require.ElementsMatch(t, []int64{1, 2, 3}, receivedIDs) + + // Every forward names the new chain, or core refuses it. + require.Equal(t, []string{"audius-mainnet-beta", "audius-mainnet-beta", "audius-mainnet-beta"}, mock.chainIDs) } // TestNewChainFlusherTrimThenSend verifies the combined trim+flush flow: diff --git a/api/server.go b/api/server.go index e1adaf92..7a792f25 100644 --- a/api/server.go +++ b/api/server.go @@ -289,7 +289,11 @@ func NewApiServer(config config.Config) *ApiServer { var newChainFlusher *NewChainFlusher if config.NewChainFlushEnabled && config.NewChainURL != "" && writePool != nil { - newChainFlusher = NewNewChainFlusher(&config, writePool, logger) + if config.NewChainID == "" { + logger.Error("new chain flusher not started: newChainId is not set, and the new chain refuses forwards without a chain ID") + } else { + newChainFlusher = NewNewChainFlusher(&config, writePool, logger) + } } app := &ApiServer{ diff --git a/config/config.go b/config/config.go index 2a5002bc..37c21373 100644 --- a/config/config.go +++ b/config/config.go @@ -89,6 +89,10 @@ type Config struct { // confirmed_block < NewChainFlushFromBlock before sending — trimming rows already // covered by the backfill. // NewChainInsecureSkipVerify disables TLS verification for the new chain endpoint (e.g. localstack). + // NewChainID is the chain the flusher stamps on every ForwardTransaction. + // Core refuses a forward whose chain ID differs from its own, and on the + // new chain also one with none, so the flusher does not start without it. + // Defaults per environment; newChainId overrides. // PlayRoutingHosts, when non-empty, are tried first when resolving a track // stream URL. Plays are recorded by whichever node serves the audio, never // through the relay, so this is what decides which chain a play lands on. @@ -105,6 +109,7 @@ type Config struct { PlayRoutingHosts []string NewChainURL string + NewChainID string NewChainQueueEnabled bool NewChainFlushEnabled bool NewChainFlushFromBlock int64 @@ -207,6 +212,7 @@ func init() { Cfg.Rewards = core_config.MakeRewards(core_config.DevClaimAuthorities, core_config.DevRewardExtensions) Cfg.AudiusdURL = "https://node1.oap.devnet" Cfg.ChainId = "openaudio-devnet" + Cfg.NewChainID = "openaudio-devnet" Cfg.SolanaIndexerWorkers = 1 Cfg.DeadNodes = []string{} Cfg.BlacklistedNodes = []string{} @@ -306,6 +312,7 @@ func init() { Cfg.Rewards = core_config.MakeRewards(core_config.ProdClaimAuthorities, core_config.ProdRewardExtensions) Cfg.AudiusdURL = "rpc.audius.co" Cfg.ChainId = "audius-mainnet-alpha-beta" + Cfg.NewChainID = "audius-mainnet-beta" Cfg.AudiusdChainID = core_config.ProdAcdcChainID Cfg.AudiusdEntityManagerAddress = core_config.ProdAcdcAddress Cfg.AudiusAppUrl = "https://audius.co" @@ -398,6 +405,9 @@ func init() { // Genesis migration dual-write queue Cfg.NewChainURL = os.Getenv("newChainUrl") + if v := os.Getenv("newChainId"); v != "" { + Cfg.NewChainID = v + } Cfg.NewChainQueueEnabled = os.Getenv("newChainQueueEnabled") == "true" Cfg.NewChainFlushEnabled = os.Getenv("newChainFlushEnabled") == "true" Cfg.NewChainInsecureSkipVerify = os.Getenv("newChainInsecureSkipVerify") == "true" diff --git a/go.mod b/go.mod index 2a675caa..26966b6b 100644 --- a/go.mod +++ b/go.mod @@ -6,7 +6,7 @@ require ( connectrpc.com/connect v1.18.1 github.com/AlecAivazis/survey/v2 v2.3.7 github.com/Doist/unfurlist v0.0.0-20250409100812-515f2735f8e5 - github.com/OpenAudio/go-openaudio v1.8.2-0.20260727214803-1d9f69772e87 + github.com/OpenAudio/go-openaudio v1.12.2-0.20261023065237-8388df0b477c github.com/OpenAudio/go-openaudio/pkg/etl v1.7.1-0.20260818221412-81f4af523801 github.com/aquasecurity/esquery v0.2.0 github.com/axiomhq/axiom-go v0.23.0 diff --git a/go.sum b/go.sum index aa6907d2..08ec98cf 100644 --- a/go.sum +++ b/go.sum @@ -18,8 +18,8 @@ github.com/Netflix/go-expect v0.0.0-20220104043353-73e0943537d2 h1:+vx7roKuyA63n github.com/Netflix/go-expect v0.0.0-20220104043353-73e0943537d2/go.mod h1:HBCaDeC1lPdgDeDbhX8XFpy1jqjK0IBG8W5K+xYqA0w= github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5 h1:TngWCqHvy9oXAN6lEVMRuU21PR1EtLVZJmdB18Gu3Rw= github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5/go.mod h1:lmUJ/7eu/Q8D7ML55dXQrVaamCz2vxCfdQBasLZfHKk= -github.com/OpenAudio/go-openaudio v1.8.2-0.20260727214803-1d9f69772e87 h1:HiLw4qrUkVWRADJjGsdHLlFMdJjhU5WCVNAfeKfww4s= -github.com/OpenAudio/go-openaudio v1.8.2-0.20260727214803-1d9f69772e87/go.mod h1:lLRvUF5oWkxOyZx8rp/ecqxuMo3yzPvuvJbLSfvxguQ= +github.com/OpenAudio/go-openaudio v1.12.2-0.20261023065237-8388df0b477c h1:saMd0Oljbar3hzuG5YA2NtYt76CbBj49EAGxevb6FTo= +github.com/OpenAudio/go-openaudio v1.12.2-0.20261023065237-8388df0b477c/go.mod h1:lLRvUF5oWkxOyZx8rp/ecqxuMo3yzPvuvJbLSfvxguQ= github.com/OpenAudio/go-openaudio/pkg/etl v1.7.1-0.20260818221412-81f4af523801 h1:lYrbDv3cte0cW5AaT5oZ+u5cLPvy9pO7S05v5uwKVHQ= github.com/OpenAudio/go-openaudio/pkg/etl v1.7.1-0.20260818221412-81f4af523801/go.mod h1:z7X/5RziXEpASGTz7tD+P3pg4W5iCA1cKW7ogw8GShY= github.com/StackExchange/wmi v1.2.1 h1:VIkavFPXSjcnS+O8yTq7NI32k0R5Aj+v39y29VYDOSA=