fix: scope offloaded node status query to the page. Fixes #16611

persist/sqldb/explosive_offload_node_status_repo.go

@@ -26,7 +26,7 @@ func (n *explosiveOffloadNodeStatusRepo) Get(context.Context, string, string) (w
 	return nil, ErrOffloadNotSupported
 }
 
-func (n *explosiveOffloadNodeStatusRepo) List(context.Context, string) (map[UUIDVersion]wfv1.Nodes, error) {
+func (n *explosiveOffloadNodeStatusRepo) List(context.Context, string, []UUIDVersion) (map[UUIDVersion]wfv1.Nodes, error) {
 	return nil, ErrOffloadNotSupported
 }
 

persist/sqldb/offload_node_status_repo.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"hash/fnv"
+	"slices"
 	"strings"
 	"time"
 
@@ -16,8 +17,6 @@ import (
 	"github.com/argoproj/argo-workflows/v4/util/sqldb"
 )
 
-const OffloadNodeStatusDisabled = "Workflow has offloaded nodes, but offloading has been disabled"
-
 type UUIDVersion struct {
 	UID     string `db:"uid"`
 	Version string `db:"version"`
@@ -26,7 +25,7 @@ type UUIDVersion struct {
 type OffloadNodeStatusRepo interface {
 	Save(ctx context.Context, uid, namespace string, nodes wfv1.Nodes) (string, error)
 	Get(ctx context.Context, uid, version string) (wfv1.Nodes, error)
-	List(ctx context.Context, namespace string) (map[UUIDVersion]wfv1.Nodes, error)
+	List(ctx context.Context, namespace string, keys []UUIDVersion) (map[UUIDVersion]wfv1.Nodes, error)
 	ListOldOffloads(ctx context.Context, namespace string) (map[string][]string, error)
 	Delete(ctx context.Context, uid, version string) error
 	IsEnabled() bool
@@ -149,34 +148,59 @@ func (wdc *nodeOffloadRepo) Get(ctx context.Context, uid, version string) (wfv1.
 	return nodes, nil
 }
 
-func (wdc *nodeOffloadRepo) List(ctx context.Context, namespace string) (map[UUIDVersion]wfv1.Nodes, error) {
-	wdc.log.WithFields(logging.Fields{"namespace": namespace}).Debug(ctx, "Listing offloaded nodes")
-	var res map[UUIDVersion]wfv1.Nodes
-	err := wdc.sessionProxy.With(ctx, func(s db.Session) error {
-		var records []nodesRecord
-		err := s.SQL().
-			Select("uid", "version", "nodes").
-			From(wdc.tableName).
-			Where(db.Cond{"clustername": wdc.clusterName}).
-			And(namespaceEqual(namespace)).
-			All(&records)
-		if err != nil {
-			return err
-		}
+// offloadListBatchSize caps how many keys go into a single query. MySQL and Postgres allow
+// 65535 placeholders per prepared statement and each pair costs two, so this leaves plenty of
+// headroom. A ceiling is needed because the caller decides the key count: ListWorkflows treats
+// an unset limit as "the whole namespace", and `argo list` defaults --chunk-size to 0.
+const offloadListBatchSize = 1000
 
-		res = make(map[UUIDVersion]wfv1.Nodes)
-		for _, r := range records {
-			nodes := &wfv1.Nodes{}
-			err = json.Unmarshal([]byte(r.Nodes), nodes)
+// uuidVersionIn matches exactly the given (uid, version) pairs. Written as OR-of-ANDs rather
+// than a row value `(uid, version) IN ((?,?),...)` so that it behaves the same on MySQL,
+// MariaDB and Postgres.
+func uuidVersionIn(keys []UUIDVersion) db.LogicalExpr {
+	conds := make([]db.LogicalExpr, len(keys))
+	for i, key := range keys {
+		conds[i] = db.And(db.Cond{"uid": key.UID}, db.Cond{"version": key.Version})
+	}
+	return db.Or(conds...)
+}
+
+// List returns the offloaded nodes for the given keys only.
+func (wdc *nodeOffloadRepo) List(ctx context.Context, namespace string, keys []UUIDVersion) (map[UUIDVersion]wfv1.Nodes, error) {
+	// This is not merely an optimisation: db.Or() with no arguments is an empty condition,
+	// so the query would silently widen back to every nodes blob in the namespace.
+	if len(keys) == 0 {
+		return map[UUIDVersion]wfv1.Nodes{}, nil
+	}
+	wdc.log.WithFields(logging.Fields{"namespace": namespace, "keys": len(keys)}).Debug(ctx, "Listing offloaded nodes")
+	res := make(map[UUIDVersion]wfv1.Nodes)
+	for batch := range slices.Chunk(keys, offloadListBatchSize) {
+		err := wdc.sessionProxy.With(ctx, func(s db.Session) error {
+			var records []nodesRecord
+			err := s.SQL().
+				Select("uid", "version", "nodes").
+				From(wdc.tableName).
+				Where(db.Cond{"clustername": wdc.clusterName}).
+				And(namespaceEqual(namespace)).
+				And(uuidVersionIn(batch)).
+				All(&records)
 			if err != nil {
 				return err
 			}
-			res[UUIDVersion{UID: r.UID, Version: r.Version}] = *nodes
+
+			for _, r := range records {
+				nodes := &wfv1.Nodes{}
+				err = json.Unmarshal([]byte(r.Nodes), nodes)
+				if err != nil {
+					return err
+				}
+				res[UUIDVersion{UID: r.UID, Version: r.Version}] = *nodes
+			}
+			return nil
+		})
+		if err != nil {
+			return nil, err
 		}
-		return nil
-	})
-	if err != nil {
-		return nil, err
 	}
 	return res, nil
 }

persist/sqldb/offload_node_status_repo_mysql_test.go

@@ -0,0 +1,72 @@
+//go:build !windows
+
+package sqldb
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	wfv1 "github.com/argoproj/argo-workflows/v4/pkg/apis/workflow/v1alpha1"
+	"github.com/argoproj/argo-workflows/v4/util/logging"
+	usqldb "github.com/argoproj/argo-workflows/v4/util/sqldb"
+)
+
+// setupMySQLOffloadTest starts a MySQL or MariaDB container and returns an offload repository.
+func setupMySQLOffloadTest(ctx context.Context, t *testing.T, v usqldb.MySQLVariant) OffloadNodeStatusRepo {
+	t.Helper()
+	repo, err := NewOffloadNodeStatusRepo(ctx, logging.RequireLoggerFromContext(ctx), setupMySQLTest(ctx, t, v), "test", "argo_workflows")
+	require.NoError(t, err)
+	return repo
+}
+
+// saveOffload writes a node status and returns the version it was stored under.
+func saveOffload(ctx context.Context, t *testing.T, repo OffloadNodeStatusRepo, uid, nodeName string) string {
+	t.Helper()
+	version, err := repo.Save(ctx, uid, "argo", wfv1.Nodes{nodeName: wfv1.NodeStatus{ID: nodeName}})
+	require.NoError(t, err)
+	return version
+}
+
+// TestMySQLListOnlyReturnsRequestedKeys covers the behaviour the list path depends on: the
+// query matches whole (uid, version) pairs, so a caller that needs one page of workflows does
+// not pull every offloaded blob in the namespace.
+//
+// Save deliberately leaves superseded rows behind for the garbage collector, so uid-a and uid-b
+// each have two versions here and only one version of each is requested.
+//
+// The version is a hash of the node contents, so two workflows only share a version value when
+// they store identical nodes. That is arranged deliberately: without it every version value is
+// unique to one uid, and a `uid IN (...) AND version IN (...)` cross-product happens to return
+// the right rows anyway. With shared version values the cross-product matches all four rows,
+// while matching whole pairs returns two.
+func TestMySQLListOnlyReturnsRequestedKeys(t *testing.T) {
+	for name, variant := range usqldb.MySQLVariants {
+		t.Run(name, func(t *testing.T) {
+			ctx := logging.TestContext(t.Context())
+			repo := setupMySQLOffloadTest(ctx, t, variant)
+
+			versionX := saveOffload(ctx, t, repo, "uid-a", "node-x")
+			versionY := saveOffload(ctx, t, repo, "uid-a", "node-y")
+			require.NotEqual(t, versionX, versionY, "different nodes must produce different versions")
+			require.Equal(t, versionX, saveOffload(ctx, t, repo, "uid-b", "node-x"), "identical nodes must share a version")
+			require.Equal(t, versionY, saveOffload(ctx, t, repo, "uid-b", "node-y"), "identical nodes must share a version")
+			unwanted := UUIDVersion{UID: "uid-c", Version: saveOffload(ctx, t, repo, "uid-c", "node-c")}
+
+			wantedA := UUIDVersion{UID: "uid-a", Version: versionX}
+			wantedB := UUIDVersion{UID: "uid-b", Version: versionY}
+
+			got, err := repo.List(ctx, "argo", []UUIDVersion{wantedA, wantedB})
+			require.NoError(t, err)
+
+			assert.Len(t, got, 2)
+			assert.Contains(t, got, wantedA)
+			assert.Contains(t, got, wantedB)
+			assert.NotContains(t, got, UUIDVersion{UID: "uid-a", Version: versionY}, "a version that was not asked for must not come back")
+			assert.NotContains(t, got, UUIDVersion{UID: "uid-b", Version: versionX}, "a version that was not asked for must not come back")
+			assert.NotContains(t, got, unwanted, "a uid that was not asked for must not come back")
+		})
+	}
+}

persist/sqldb/offload_node_status_repo_test.go

@@ -7,8 +7,18 @@ import (
 	"github.com/stretchr/testify/require"
 
 	wfv1 "github.com/argoproj/argo-workflows/v4/pkg/apis/workflow/v1alpha1"
+	"github.com/argoproj/argo-workflows/v4/util/logging"
 )
 
+// Test_ListWithoutKeys guards the empty-keys early return. Without it the generated condition
+// is empty, which would widen the query back to the whole namespace. A zero-value repo has no
+// session, so anything past the guard panics rather than quietly querying.
+func Test_ListWithoutKeys(t *testing.T) {
+	got, err := (&nodeOffloadRepo{}).List(logging.TestContext(t.Context()), "argo", nil)
+	require.NoError(t, err)
+	assert.Empty(t, got)
+}
+
 func Test_nodeStatusVersion(t *testing.T) {
 	t.Run("Empty", func(t *testing.T) {
 		marshalled, version, err := nodeStatusVersion(nil)

persist/sqldb/workflow_archive_mysql_test.go

@@ -27,6 +27,13 @@ import (
 // setupMySQLArchiveTest starts a MySQL or MariaDB container, runs migrations, and returns a WorkflowArchive.
 func setupMySQLArchiveTest(ctx context.Context, t *testing.T, v usqldb.MySQLVariant) WorkflowArchive {
 	t.Helper()
+	return NewWorkflowArchive(setupMySQLTest(ctx, t, v), "test", "", instanceid.NewService(""))
+}
+
+// setupMySQLTest starts a MySQL or MariaDB container, runs migrations, and returns a session
+// proxy the caller can build any repository on.
+func setupMySQLTest(ctx context.Context, t *testing.T, v usqldb.MySQLVariant) *usqldb.SessionProxy {
+	t.Helper()
 
 	c, err := testmysql.Run(ctx,
 		v.Image,
@@ -73,7 +80,7 @@ func setupMySQLArchiveTest(ctx context.Context, t *testing.T, v usqldb.MySQLVari
 
 	t.Cleanup(func() { proxy.Close() })
 
-	return NewWorkflowArchive(proxy, "test", "", instanceid.NewService(""))
+	return proxy
 }
 
 // TestMySQLListWorkflows verifies that JSON_EXTRACT/JSON_UNQUOTE queries in

server/workflow/workflow_server.go

@@ -5,6 +5,8 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"maps"
+	"slices"
 	"sort"
 	"sync"
 	"time"
@@ -285,19 +287,22 @@ func (s *workflowServer) ListWorkflows(ctx context.Context, req *workflowpkg.Wor
 	}
 
 	cleaner := fields.NewCleaner(req.Fields)
-	logger := logging.RequireLoggerFromContext(ctx)
 	if s.offloadNodeStatusRepo.IsEnabled() && !cleaner.WillExclude("items.status.nodes") {
-		offloadedNodes, err := s.offloadNodeStatusRepo.List(ctx, req.Namespace)
-		if err != nil {
-			return nil, sutils.ToStatusError(err, codes.Internal)
-		}
+		// This page is already resolved, so we know exactly which offloaded rows we need.
+		offloaded := map[int]sqldb.UUIDVersion{}
 		for i, wf := range wfs {
 			if wf.Status.IsOffloadNodeStatus() {
-				if s.offloadNodeStatusRepo.IsEnabled() {
-					wfs[i].Status.Nodes = offloadedNodes[sqldb.UUIDVersion{UID: string(wf.UID), Version: wf.GetOffloadNodeStatusVersion()}]
-				} else {
-					logger.WithFields(logging.Fields{"namespace": wf.Namespace, "name": wf.Name}).Warn(ctx, sqldb.OffloadNodeStatusDisabled)
-				}
+				offloaded[i] = sqldb.UUIDVersion{UID: string(wf.UID), Version: wf.GetOffloadNodeStatusVersion()}
+			}
+		}
+		// Nothing on this page is offloaded, so there is nothing to fetch.
+		if len(offloaded) > 0 {
+			offloadedNodes, err := s.offloadNodeStatusRepo.List(ctx, req.Namespace, slices.Collect(maps.Values(offloaded)))
+			if err != nil {
+				return nil, sutils.ToStatusError(err, codes.Internal)
+			}
+			for i, key := range offloaded {
+				wfs[i].Status.Nodes = offloadedNodes[key]
 			}
 		}
 	}