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=