feat(api): route stream resolution through named hosts during the chain migration

api/play_routing.go

@@ -0,0 +1,89 @@
+package api
+
+import (
+	"net/url"
+	"strings"
+
+	"api.audius.co/api/dbv1"
+)
+
+// withPlayRoutingHosts returns a copy of link whose candidate hosts are the
+// configured routing hosts first, then the link's own url and mirrors.
+//
+// A play is recorded by whichever node serves the audio -- logTrackListen runs
+// at the top of mediorum's serveBlob, before it 307s to storage -- and plays
+// never travel through the relay. So the host in the URL the API hands back is
+// what decides which chain the play lands on.
+//
+// During the genesis migration the fleet is split across two chains for days.
+// An already-migrated node writes its plays to the new chain while the indexer
+// is still reading the old one, and those plays are indexed by nobody. Naming
+// hosts that stay on the old chain keeps every play on the chain the indexer is
+// actually reading. Cleared at the cutover, after which plays follow the node
+// serving them again. See cmd/genesis-writer/ROLLOUT.md, Runbook steps 5 and 13.
+//
+// The original url and mirrors are kept as fallbacks rather than replaced. The
+// routing hosts are store-all nodes and hold essentially everything, but
+// replication of a fresh upload is not instant, and a track they do not have
+// yet must still be streamable. tryFindWorkingUrl probes in order, so a routing
+// host that cannot serve costs one request and falls through.
+func withPlayRoutingHosts(link *dbv1.MediaLink, hosts []string) *dbv1.MediaLink {
+	if link == nil || len(hosts) == 0 {
+		return link
+	}
+
+	primary, err := url.Parse(link.Url)
+	if err != nil {
+		return link
+	}
+
+	routed := &dbv1.MediaLink{}
+	seen := make(map[string]bool, len(hosts)+len(link.Mirrors)+1)
+
+	// Host-only comparison: mirrors are recorded as hosts, and the routing hosts
+	// are configured as hosts, so normalising to that avoids listing the same
+	// node twice under different spellings.
+	add := func(raw string) {
+		h := hostOf(raw)
+		if h == "" || seen[h] {
+			return
+		}
+		seen[h] = true
+		if routed.Url == "" {
+			u := *primary
+			u.Host = h
+			routed.Url = u.String()
+			return
+		}
+		routed.Mirrors = append(routed.Mirrors, h)
+	}
+
+	for _, h := range hosts {
+		add(h)
+	}
+	add(link.Url)
+	for _, m := range link.Mirrors {
+		add(m)
+	}
+
+	if routed.Url == "" {
+		return link
+	}
+	return routed
+}
+
+// hostOf accepts either a bare host or a full URL and returns the host.
+func hostOf(s string) string {
+	s = strings.TrimSpace(s)
+	if s == "" {
+		return ""
+	}
+	if !strings.Contains(s, "//") {
+		s = "https://" + s
+	}
+	u, err := url.Parse(s)
+	if err != nil {
+		return ""
+	}
+	return u.Host
+}

api/play_routing_test.go

@@ -0,0 +1,89 @@
+package api
+
+import (
+	"net/url"
+	"testing"
+
+	"api.audius.co/api/dbv1"
+	"github.com/stretchr/testify/require"
+)
+
+const streamPath = "/tracks/cidstream/abc?signature=sig"
+
+func hostsOf(t *testing.T, link *dbv1.MediaLink) []string {
+	t.Helper()
+	u, err := url.Parse(link.Url)
+	require.NoError(t, err)
+	return append([]string{u.Host}, link.Mirrors...)
+}
+
+// Unconfigured, this must be exactly inert -- it ships ahead of the migration
+// and sits dormant in production until someone sets the env var.
+func TestPlayRoutingIsInertWhenUnconfigured(t *testing.T) {
+	link := &dbv1.MediaLink{Url: "https://node-a.example" + streamPath, Mirrors: []string{"node-b.example"}}
+	require.Same(t, link, withPlayRoutingHosts(link, nil))
+	require.Same(t, link, withPlayRoutingHosts(link, []string{}))
+	require.Nil(t, withPlayRoutingHosts(nil, []string{"x.example"}))
+}
+
+// Routing hosts go first, because tryFindWorkingUrl probes in order and the
+// first host that can serve is the one that records the play.
+func TestPlayRoutingHostsAreTriedFirst(t *testing.T) {
+	link := &dbv1.MediaLink{Url: "https://node-a.example" + streamPath, Mirrors: []string{"node-b.example"}}
+	routed := withPlayRoutingHosts(link, []string{"creatornode.audius.co", "v.monophonic.digital"})
+
+	require.Equal(t,
+		[]string{"creatornode.audius.co", "v.monophonic.digital", "node-a.example", "node-b.example"},
+		hostsOf(t, routed))
+}
+
+// The original hosts stay as fallbacks. Store-all nodes hold nearly everything,
+// but a freshly uploaded track may not have replicated yet, and it still has to
+// be streamable -- just with the play landing on whichever node serves it.
+func TestPlayRoutingKeepsOriginalHostsAsFallback(t *testing.T) {
+	link := &dbv1.MediaLink{Url: "https://node-a.example" + streamPath, Mirrors: []string{"node-b.example"}}
+	routed := withPlayRoutingHosts(link, []string{"creatornode.audius.co"})
+
+	require.Contains(t, hostsOf(t, routed), "node-a.example")
+	require.Contains(t, hostsOf(t, routed), "node-b.example")
+}
+
+// The path and query -- including the signature mediorum parses to attribute the
+// listen -- must survive the host rewrite, or the play is recorded against the
+// wrong user or not at all.
+func TestPlayRoutingPreservesPathAndSignature(t *testing.T) {
+	link := &dbv1.MediaLink{Url: "https://node-a.example" + streamPath}
+	routed := withPlayRoutingHosts(link, []string{"creatornode.audius.co"})
+
+	u, err := url.Parse(routed.Url)
+	require.NoError(t, err)
+	require.Equal(t, "creatornode.audius.co", u.Host)
+	require.Equal(t, "/tracks/cidstream/abc", u.Path)
+	require.Equal(t, "sig", u.Query().Get("signature"))
+}
+
+// A routing host that is already the primary or a mirror must not be probed
+// twice; duplicates would waste a request and could double-count if one of them
+// ever lost its skip_play_count.
+func TestPlayRoutingDeduplicatesHosts(t *testing.T) {
+	link := &dbv1.MediaLink{Url: "https://creatornode.audius.co" + streamPath, Mirrors: []string{"node-b.example"}}
+	routed := withPlayRoutingHosts(link, []string{"creatornode.audius.co", "node-b.example"})
+
+	require.Equal(t, []string{"creatornode.audius.co", "node-b.example"}, hostsOf(t, routed))
+}
+
+// Hosts may be configured bare or as full URLs; both must normalise to the same
+// thing so a scheme in the env var does not silently create a duplicate.
+func TestPlayRoutingAcceptsBareHostsAndUrls(t *testing.T) {
+	link := &dbv1.MediaLink{Url: "https://node-a.example" + streamPath}
+	bare := withPlayRoutingHosts(link, []string{"creatornode.audius.co"})
+	full := withPlayRoutingHosts(link, []string{"https://creatornode.audius.co"})
+	require.Equal(t, hostsOf(t, bare), hostsOf(t, full))
+}
+
+// An unparseable link is returned untouched rather than dropped: streaming
+// should degrade to current behaviour, never fail, because of this feature.
+func TestPlayRoutingLeavesUnparseableLinkAlone(t *testing.T) {
+	link := &dbv1.MediaLink{Url: "://not a url"}
+	require.Same(t, link, withPlayRoutingHosts(link, []string{"creatornode.audius.co"}))
+}

api/v1_track_stream.go

@@ -52,6 +52,11 @@ func (app *ApiServer) v1TrackStream(c *fiber.Ctx) error {
 }
 
 func (app *ApiServer) redirectToStream(c *fiber.Ctx, stream *dbv1.MediaLink) error {
+	// Temporary, for the genesis migration: prefer hosts that stay on the old
+	// chain so plays are not split across two chains while the fleet migrates.
+	// No-op when unconfigured. See withPlayRoutingHosts.
+	stream = withPlayRoutingHosts(stream, app.config.PlayRoutingHosts)
+
 	streamURL := tryFindWorkingUrl(stream)
 
 	if skipPlayCount := c.Query("skip_play_count"); skipPlayCount != "" {

config/config.go

@@ -89,6 +89,18 @@ 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).
+	// 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.
+	//
+	// It exists for the genesis migration: while the fleet is split across two
+	// chains, an already-migrated node writes its plays to the new chain even
+	// though the indexer is still reading the old one, and every play in that
+	// window goes to a chain nobody reads. Pointing this at nodes that stay on
+	// the old chain keeps plays where the indexer is. Cleared at the cutover.
+	// See cmd/genesis-writer/ROLLOUT.md, Runbook steps 5 and 13.
+	PlayRoutingHosts []string
+
 	NewChainURL                string
 	NewChainQueueEnabled       bool
 	NewChainFlushEnabled       bool
@@ -362,6 +374,15 @@ func init() {
 		Cfg.FeaturedAudienceUserID = int32(parsed)
 	}
 
+	// Genesis migration: temporary play routing (see the struct field).
+	if v := strings.TrimSpace(os.Getenv("playRoutingHosts")); v != "" {
+		for _, h := range strings.Split(v, ",") {
+			if h = strings.TrimSpace(h); h != "" {
+				Cfg.PlayRoutingHosts = append(Cfg.PlayRoutingHosts, h)
+			}
+		}
+	}
+
 	// Genesis migration dual-write queue
 	Cfg.NewChainURL = os.Getenv("newChainUrl")
 	Cfg.NewChainQueueEnabled = os.Getenv("newChainQueueEnabled") == "true"