test(i): Add cross-version P2P test multiplier

.github/workflows/test-coverage.yml

@@ -388,6 +388,65 @@ jobs:
           coverage-artifact-name: "coverage_signed_docs"
           coverage-path: coverage.txt
 
+  # These jobs run the networked tests with one node on an older release, to catch
+  # changes that break compatibility with it. Both directions are run because they
+  # fail differently: old-source has the older node sending, new-source receiving.
+  #
+  # The release binary is a public download over HTTPS, so no token is needed.
+  test-coverage-cross-version-old-source:
+    name: Test coverage cross version old source job
+
+    runs-on: runs-on=${{ github.run_id }}-${{ github.run_attempt }}-${{ strategy.job-index }}/\
+      spot=pco/cpu=16+32/family=c6*+c7*/disk=large/extras=s3-cache
+
+    env:
+      DEFRA_MULTIPLIERS: cross-version-old-source
+
+    steps:
+      - name: Enable RunsOn action
+        uses: runs-on/action@v2
+        with:
+          metrics: cpu,network,memory,disk,io
+
+      - name: Checkout code into the directory
+        uses: actions/checkout@v6
+
+      - name: Setup defradb
+        uses: ./.github/composites/setup-defradb
+
+      - name: Test coverage & save coverage report in an artifact
+        uses: ./.github/composites/test-coverage-with-artifact
+        with:
+          coverage-artifact-name: "coverage_cross_version_old_source"
+          coverage-path: coverage.txt
+
+  test-coverage-cross-version-new-source:
+    name: Test coverage cross version new source job
+
+    runs-on: runs-on=${{ github.run_id }}-${{ github.run_attempt }}-${{ strategy.job-index }}/\
+      spot=pco/cpu=16+32/family=c6*+c7*/disk=large/extras=s3-cache
+
+    env:
+      DEFRA_MULTIPLIERS: cross-version-new-source
+
+    steps:
+      - name: Enable RunsOn action
+        uses: runs-on/action@v2
+        with:
+          metrics: cpu,network,memory,disk,io
+
+      - name: Checkout code into the directory
+        uses: actions/checkout@v6
+
+      - name: Setup defradb
+        uses: ./.github/composites/setup-defradb
+
+      - name: Test coverage & save coverage report in an artifact
+        uses: ./.github/composites/test-coverage-with-artifact
+        with:
+          coverage-artifact-name: "coverage_cross_version_new_source"
+          coverage-path: coverage.txt
+
   # This job tests the leveldb datastore.
   test-coverage-leveldb:
     name: Test coverage leveldb job
@@ -431,6 +490,8 @@ jobs:
       - test-coverage-js              #  1 test(s)
       - test-coverage-secondary-index #  1 test(s)
       - test-coverage-signed-docs     #  1 test(s)
+      - test-coverage-cross-version-old-source  #  1 test(s)
+      - test-coverage-cross-version-new-source  #  1 test(s)
       - test-coverage-leveldb         #  1 test(s)
 
     # Important to know:

tests/action/acp_dac_config.go

@@ -0,0 +1,33 @@
+// Copyright 2026 Democratized Data Foundation
+//
+// This file is part of the DefraDB test suite.
+//
+// The DefraDB test suite is licensed under either:
+//
+//   (1) GNU Affero General Public License v3
+//   (2) Business Source License 1.1
+//
+// See tests/LICENSE for details.
+
+package action
+
+import (
+	"os"
+
+	"github.com/sourcenetwork/defradb/tests/state"
+)
+
+const documentACPTypeEnvName = "DEFRA_DOCUMENT_ACP_TYPE"
+
+// DocumentACPType is the document ACP implementation under test.
+//
+// Node setup and the test harness both read this, so it is resolved once here
+// rather than copied into each package.
+var DocumentACPType state.DocumentACPType
+
+func init() {
+	DocumentACPType = state.DocumentACPType(os.Getenv(documentACPTypeEnvName))
+	if DocumentACPType == "" {
+		DocumentACPType = state.LocalDocumentACPType
+	}
+}

tests/action/assert_request.go

@@ -156,7 +156,7 @@ func assertRequestResults(
 		default:
 			assertResultsEqual(
 				s.T,
-				s.ClientType,
+				clientTypeForNode(s, nodeID),
 				expect,
 				actual,
 				fmt.Sprintf("node: %v, path: %s", nodeID, stack),
@@ -273,14 +273,14 @@ func assertRequestResultDoc(
 			if ordered {
 				assertResultsEqual(
 					s.T,
-					s.ClientType,
+					clientTypeForNode(s, nodeID),
 					expectedDocID,
 					actualValue,
 					fmt.Sprintf("node: %v, path: %s", nodeID, stack),
 				)
 			} else {
 				ok := isResultsEqual(
-					s.ClientType,
+					clientTypeForNode(s, nodeID),
 					expectedDocID,
 					actualValue,
 				)
@@ -323,14 +323,14 @@ func assertRequestResultDoc(
 			if ordered {
 				assertResultsEqual(
 					s.T,
-					s.ClientType,
+					clientTypeForNode(s, nodeID),
 					expectedValue,
 					actualValue,
 					fmt.Sprintf("node: %v, path: %s", nodeID, stack),
 				)
 			} else {
 				ok := isResultsEqual(
-					s.ClientType,
+					clientTypeForNode(s, nodeID),
 					expectedValue,
 					actualValue,
 				)

tests/action/eventually.go

@@ -0,0 +1,163 @@
+// Copyright 2026 Democratized Data Foundation
+//
+// This file is part of the DefraDB test suite.
+//
+// The DefraDB test suite is licensed under either:
+//
+//   (1) GNU Affero General Public License v3
+//   (2) Business Source License 1.1
+//
+// See tests/LICENSE for details.
+
+package action
+
+import (
+	"fmt"
+	"testing"
+	"time"
+)
+
+const (
+	// eventuallyTimeout is how long [Eventually] retries before giving up.
+	eventuallyTimeout = 20 * time.Second
+	// eventuallyInterval is how long [Eventually] waits between attempts.
+	eventuallyInterval = 100 * time.Millisecond
+)
+
+// Eventually runs another action until it stops failing.
+//
+// Use it when the harness cannot tell that something has finished and the test
+// has to keep asking. The clearest case is a node running an older release: it
+// can hold a document written against a schema it has never seen, but it cannot
+// report the commit, so there is no signal to wait for.
+//
+// The wrapped action's assertions are captured rather than failing the test, so
+// a failed attempt is just a retry. The final attempt's failure is reported as
+// this action's failure.
+type Eventually struct {
+	stateful
+
+	// Action is retried until it passes or the timeout is reached.
+	Action Action
+
+	// Timeout overrides the default retry window.
+	Timeout time.Duration
+}
+
+var _ Action = (*Eventually)(nil)
+var _ Stateful = (*Eventually)(nil)
+
+func (a *Eventually) Execute() {
+	timeout := a.Timeout
+	if timeout == 0 {
+		timeout = eventuallyTimeout
+	}
+
+	realT := a.s.T
+	// Restore with a defer: a real panic from the nested action skips past the
+	// assignment below, which would leave the state pointing at a recorder that
+	// nothing reads, silently swallowing later failures.
+	defer func() { a.s.T = realT }()
+	deadline := time.Now().Add(timeout)
+
+	var lastErr string
+	for {
+		recorder := &recordingT{TB: realT}
+		a.s.T = recorder
+		if stateful, ok := a.Action.(Stateful); ok {
+			stateful.SetState(a.s)
+		}
+		failed := a.attempt(recorder)
+		a.s.T = realT
+
+		if !failed {
+			return
+		}
+		lastErr = recorder.message
+
+		if time.Now().After(deadline) {
+			a.s.T.Errorf("action did not pass within %s: %s", timeout, lastErr)
+			a.s.T.FailNow()
+			return
+		}
+		time.Sleep(eventuallyInterval)
+	}
+}
+
+// attempt runs the action once, reporting whether it failed rather than failing
+// the test.
+func (a *Eventually) attempt(recorder *recordingT) (failed bool) {
+	defer func() {
+		r := recover()
+		if r == nil {
+			return
+		}
+		if _, ok := r.(attemptFailure); ok {
+			failed = true
+			return
+		}
+		// Anything else is a real panic and belongs to the caller.
+		panic(r)
+	}()
+
+	a.Action.Execute()
+	return recorder.failed
+}
+
+// attemptFailure marks an attempt that ended early because an assertion failed.
+//
+// The failure is raised as a panic rather than by ending the goroutine, so the
+// retry loop can recover from it in place. This mirrors how the flake retry
+// helper in the integration package handles the same problem.
+type attemptFailure struct{}
+
+// recordingT captures assertion failures instead of failing the test.
+type recordingT struct {
+	testing.TB
+
+	failed  bool
+	message string
+}
+
+func (t *recordingT) Errorf(format string, args ...any) {
+	t.fail(fmt.Sprintf(format, args...))
+}
+
+func (t *recordingT) Error(args ...any) {
+	t.fail(fmt.Sprint(args...))
+}
+
+func (t *recordingT) Fatal(args ...any) {
+	t.fail(fmt.Sprint(args...))
+	t.FailNow()
+}
+
+func (t *recordingT) Fatalf(format string, args ...any) {
+	t.fail(fmt.Sprintf(format, args...))
+	t.FailNow()
+}
+
+func (t *recordingT) Fail() {
+	t.fail("")
+}
+
+func (t *recordingT) FailNow() {
+	t.failed = true
+	panic(attemptFailure{})
+}
+
+func (t *recordingT) Failed() bool {
+	return t.failed
+}
+
+func (t *recordingT) fail(message string) {
+	t.failed = true
+	if message != "" {
+		t.message = message
+	}
+}
+
+// NewEventually returns an [Eventually] wrapping the given action.
+func NewEventually(action Action) *Eventually {
+	return &Eventually{Action: action}
+}