fix(supernode): preserve audit identities across account migration (companion to chain #199)

pkg/lumera/modules/audit/assignment.go

@@ -0,0 +1,76 @@
+package audit
+
+import (
+	"fmt"
+	"strings"
+
+	audittypes "github.com/LumeraProtocol/lumera/x/audit/v1/types"
+)
+
+// AssignedTarget separates the identity frozen into the epoch assignment from
+// the account that currently owns and serves that identity.
+type AssignedTarget struct {
+	LogicalAccount string
+	CurrentAccount string
+}
+
+// AssignedTargets is the validated, continuity-aware projection of the chain
+// response. Logical accounts are used in reports and deterministic transcripts;
+// current accounts are used only for live network routing.
+type AssignedTargets struct {
+	EpochID           uint64
+	ReporterAccount   string
+	RequiredOpenPorts []uint32
+	Targets           []AssignedTarget
+}
+
+// ResolveAssignedTargets validates the chain-provided identity mappings. It
+// deliberately does not consult SuperNode.PrevSupernodeAccounts: only Audit's
+// indexed lineage is authoritative for an epoch assignment.
+func ResolveAssignedTargets(resp *audittypes.QueryAssignedTargetsResponse, requestedEpoch uint64) (AssignedTargets, error) {
+	if resp == nil {
+		return AssignedTargets{}, fmt.Errorf("assigned targets response is nil")
+	}
+	if resp.EpochId != requestedEpoch {
+		return AssignedTargets{}, fmt.Errorf("assigned targets epoch mismatch: got %d, want %d", resp.EpochId, requestedEpoch)
+	}
+	reporter := strings.TrimSpace(resp.ReporterSupernodeAccount)
+	if reporter == "" {
+		return AssignedTargets{}, fmt.Errorf("assigned targets response is missing logical reporter")
+	}
+	if len(resp.TargetAccountMappings) != len(resp.TargetSupernodeAccounts) {
+		return AssignedTargets{}, fmt.Errorf("assigned targets mapping count mismatch: got %d mappings for %d targets", len(resp.TargetAccountMappings), len(resp.TargetSupernodeAccounts))
+	}
+
+	resolved := AssignedTargets{
+		EpochID:           resp.EpochId,
+		ReporterAccount:   reporter,
+		RequiredOpenPorts: append([]uint32(nil), resp.RequiredOpenPorts...),
+		Targets:           make([]AssignedTarget, len(resp.TargetSupernodeAccounts)),
+	}
+	seenLogical := make(map[string]struct{}, len(resolved.Targets))
+	seenCurrent := make(map[string]struct{}, len(resolved.Targets))
+	for i, expected := range resp.TargetSupernodeAccounts {
+		logical := strings.TrimSpace(resp.TargetAccountMappings[i].LogicalAccount)
+		current := strings.TrimSpace(resp.TargetAccountMappings[i].CurrentAccount)
+		if logical == "" || current == "" {
+			return AssignedTargets{}, fmt.Errorf("assigned target mapping %d has an empty account", i)
+		}
+		if logical != strings.TrimSpace(expected) {
+			return AssignedTargets{}, fmt.Errorf("assigned target mapping %d logical account mismatch: got %q, want %q", i, logical, expected)
+		}
+		if _, exists := seenLogical[logical]; exists {
+			return AssignedTargets{}, fmt.Errorf("assigned target mapping duplicates logical account %q", logical)
+		}
+		if logical == reporter {
+			return AssignedTargets{}, fmt.Errorf("assigned target mapping duplicates logical reporter %q", reporter)
+		}
+		if _, exists := seenCurrent[current]; exists {
+			return AssignedTargets{}, fmt.Errorf("assigned target mapping aliases current account %q", current)
+		}
+		seenLogical[logical] = struct{}{}
+		seenCurrent[current] = struct{}{}
+		resolved.Targets[i] = AssignedTarget{LogicalAccount: logical, CurrentAccount: current}
+	}
+	return resolved, nil
+}

pkg/lumera/modules/audit/assignment_test.go

@@ -0,0 +1,85 @@
+package audit
+
+import (
+	"testing"
+
+	audittypes "github.com/LumeraProtocol/lumera/x/audit/v1/types"
+	"github.com/stretchr/testify/require"
+)
+
+func assignedResponse(epoch uint64, reporter string, logical, current []string) *audittypes.QueryAssignedTargetsResponse {
+	mappings := make([]audittypes.AccountIdentityMapping, len(logical))
+	for i := range logical {
+		mappings[i] = audittypes.AccountIdentityMapping{LogicalAccount: logical[i], CurrentAccount: current[i]}
+	}
+	return &audittypes.QueryAssignedTargetsResponse{
+		EpochId:                  epoch,
+		ReporterSupernodeAccount: reporter,
+		TargetSupernodeAccounts:  append([]string(nil), logical...),
+		TargetAccountMappings:    mappings,
+		RequiredOpenPorts:        []uint32{4444, 5555},
+	}
+}
+
+func TestResolveAssignedTargetsMigrationMatrix(t *testing.T) {
+	tests := []struct {
+		name            string
+		requested       string
+		reporterLogical string
+		targetLogical   string
+		targetCurrent   string
+	}{
+		{name: "unmigrated", requested: "reporter-A", reporterLogical: "reporter-A", targetLogical: "target-A", targetCurrent: "target-A"},
+		{name: "reporter-only", requested: "reporter-B", reporterLogical: "reporter-A", targetLogical: "target-A", targetCurrent: "target-A"},
+		{name: "target-only", requested: "reporter-A", reporterLogical: "reporter-A", targetLogical: "target-A", targetCurrent: "target-B"},
+		{name: "both", requested: "reporter-B", reporterLogical: "reporter-A", targetLogical: "target-A", targetCurrent: "target-B"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			resp := assignedResponse(17, tt.reporterLogical, []string{tt.targetLogical}, []string{tt.targetCurrent})
+			got, err := ResolveAssignedTargets(resp, 17)
+			require.NoError(t, err)
+			require.Equal(t, tt.reporterLogical, got.ReporterAccount)
+			require.Equal(t, []AssignedTarget{{LogicalAccount: tt.targetLogical, CurrentAccount: tt.targetCurrent}}, got.Targets)
+			// The request account is deliberately independent: the adapter must
+			// trust the chain's epoch-logical reporter projection.
+			require.NotEmpty(t, tt.requested)
+		})
+	}
+}
+
+func TestResolveAssignedTargetsRejectsNextEpochAndMalformedMappings(t *testing.T) {
+	valid := func() *audittypes.QueryAssignedTargetsResponse {
+		return assignedResponse(21, "reporter-A", []string{"target-A", "target-C"}, []string{"target-B", "target-D"})
+	}
+	tests := []struct {
+		name   string
+		mutate func(*audittypes.QueryAssignedTargetsResponse)
+		want   string
+	}{
+		{name: "next epoch", mutate: func(r *audittypes.QueryAssignedTargetsResponse) { r.EpochId = 22 }, want: "epoch mismatch"},
+		{name: "missing reporter", mutate: func(r *audittypes.QueryAssignedTargetsResponse) { r.ReporterSupernodeAccount = " " }, want: "missing logical reporter"},
+		{name: "mapping count", mutate: func(r *audittypes.QueryAssignedTargetsResponse) {
+			r.TargetAccountMappings = r.TargetAccountMappings[:1]
+		}, want: "mapping count mismatch"},
+		{name: "mapping order", mutate: func(r *audittypes.QueryAssignedTargetsResponse) {
+			r.TargetAccountMappings[0], r.TargetAccountMappings[1] = r.TargetAccountMappings[1], r.TargetAccountMappings[0]
+		}, want: "logical account mismatch"},
+		{name: "duplicate logical", mutate: func(r *audittypes.QueryAssignedTargetsResponse) {
+			r.TargetSupernodeAccounts[1] = "target-A"
+			r.TargetAccountMappings[1].LogicalAccount = "target-A"
+		}, want: "duplicates logical"},
+		{name: "duplicate current", mutate: func(r *audittypes.QueryAssignedTargetsResponse) {
+			r.TargetAccountMappings[1].CurrentAccount = "target-B"
+		}, want: "aliases current"},
+		{name: "empty account", mutate: func(r *audittypes.QueryAssignedTargetsResponse) { r.TargetAccountMappings[0].CurrentAccount = " " }, want: "empty account"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			resp := valid()
+			tt.mutate(resp)
+			_, err := ResolveAssignedTargets(resp, 21)
+			require.ErrorContains(t, err, tt.want)
+		})
+	}
+}

supernode/host_reporter/service.go

@@ -19,6 +19,7 @@ import (
 	"github.com/LumeraProtocol/supernode/v2/pkg/logtrace"
 	"github.com/LumeraProtocol/supernode/v2/pkg/lumera"
 	"github.com/LumeraProtocol/supernode/v2/pkg/lumera/chainerrors"
+	auditmod "github.com/LumeraProtocol/supernode/v2/pkg/lumera/modules/audit"
 	"github.com/LumeraProtocol/supernode/v2/pkg/reachability"
 	statussvc "github.com/LumeraProtocol/supernode/v2/supernode/status"
 	"github.com/cosmos/cosmos-sdk/crypto/keyring"
@@ -174,24 +175,31 @@ func (s *Service) tick(ctx context.Context) {
 		return
 	}
 
-	// Idempotency: if a report exists for this epoch, do nothing.
-	if _, err := s.lumera.Audit().GetEpochReport(tickCtx, epochID, s.identity); err == nil {
+	assignResp, err := s.lumera.Audit().GetAssignedTargets(tickCtx, s.identity, epochID)
+	if err != nil || assignResp == nil {
 		return
-	} else if status.Code(err) != codes.NotFound {
+	}
+	assignment, err := auditmod.ResolveAssignedTargets(assignResp, epochID)
+	if err != nil {
+		logtrace.Warn(tickCtx, "epoch report skipped: invalid identity assignment", logtrace.Fields{"epoch_id": epochID, "error": err.Error()})
 		return
 	}
 
-	assignResp, err := s.lumera.Audit().GetAssignedTargets(tickCtx, s.identity, epochID)
-	if err != nil || assignResp == nil {
+	// Idempotency is keyed by the epoch-logical reporter, while the transaction
+	// is still signed and submitted by the configured current account.
+	if _, err := s.lumera.Audit().GetEpochReport(tickCtx, epochID, assignment.ReporterAccount); err == nil {
+		return
+	} else if status.Code(err) != codes.NotFound {
 		return
 	}
 
-	storageChallengeObservations := s.buildStorageChallengeObservations(tickCtx, epochID, assignResp.RequiredOpenPorts, assignResp.TargetSupernodeAccounts)
+	storageChallengeObservations := s.buildStorageChallengeObservations(tickCtx, epochID, assignment.RequiredOpenPorts, assignment.Targets)
 
 	var storageProofResults []*audittypes.StorageProofResult
 	proofResultProvider := s.getProofResultProvider()
 	if proofResultProvider != nil {
 		storageProofResults = proofResultProvider.CollectResults(epochID)
+		storageProofResults = compatibleProofResults(storageProofResults, assignment)
 		mode, modeOK := s.storageTruthEnforcementMode(tickCtx)
 		if modeOK && mode == audittypes.StorageTruthEnforcementMode_STORAGE_TRUTH_ENFORCEMENT_MODE_FULL {
 			// FULL mode is the only mode where the chain enforces compound
@@ -201,18 +209,18 @@ func (s *Service) tick(ctx context.Context) {
 			// drain doesn't satisfy that, we MUST skip this epoch and
 			// requeue the partial rows so the next tick can try again with
 			// a complete set.
-			complete, reason := storageProofCoverageComplete(storageProofResults, assignResp.TargetSupernodeAccounts)
+			complete, reason := storageProofCoverageComplete(storageProofResults, logicalTargetAccounts(assignment.Targets))
 			if !complete {
 				requeueProofResults(proofResultProvider, epochID, storageProofResults)
 				logtrace.Warn(tickCtx, "epoch report skipped: incomplete FULL-mode storage proof coverage", logtrace.Fields{
 					"epoch_id":         epochID,
-					"assigned_targets": len(assignResp.TargetSupernodeAccounts),
+					"assigned_targets": len(assignment.Targets),
 					"proof_results":    len(storageProofResults),
 					"reason":           reason,
 				})
 				return
 			}
-		} else if modeOK && len(assignResp.TargetSupernodeAccounts) > 0 && len(storageProofResults) == 0 {
+		} else if modeOK && len(assignment.Targets) > 0 && len(storageProofResults) == 0 {
 			// SHADOW / SOFT / UNSPECIFIED: chain accepts empty StorageProofResults
 			// (only FULL enforces compound coverage). Submitting the host /
 			// peer-observation report is mandatory regardless — withholding it
@@ -224,7 +232,7 @@ func (s *Service) tick(ctx context.Context) {
 			// scoring (LEP-6 PR286 review F1).
 			logtrace.Info(tickCtx, "epoch report: submitting in non-FULL mode with empty LEP-6 proof rows", logtrace.Fields{
 				"epoch_id":         epochID,
-				"assigned_targets": len(assignResp.TargetSupernodeAccounts),
+				"assigned_targets": len(assignment.Targets),
 				"mode":             mode.String(),
 			})
 		}
@@ -305,6 +313,36 @@ func requeueProofResults(provider ProofResultProvider, epochID uint64, results [
 	}
 }
 
+func logicalTargetAccounts(targets []auditmod.AssignedTarget) []string {
+	out := make([]string, len(targets))
+	for i := range targets {
+		out[i] = targets[i].LogicalAccount
+	}
+	return out
+}
+
+// compatibleProofResults implements the migration buffer policy. Rows built
+// with current/pre-migration identities cannot be repaired because their
+// transcript and signature cover those identities; discard them so the
+// dispatcher rebuilds them from the authoritative assignment.
+func compatibleProofResults(results []*audittypes.StorageProofResult, assignment auditmod.AssignedTargets) []*audittypes.StorageProofResult {
+	allowed := make(map[string]struct{}, len(assignment.Targets))
+	for _, target := range assignment.Targets {
+		allowed[target.LogicalAccount] = struct{}{}
+	}
+	out := make([]*audittypes.StorageProofResult, 0, len(results))
+	for _, result := range results {
+		if result == nil || result.ChallengerSupernodeAccount != assignment.ReporterAccount {
+			continue
+		}
+		if _, ok := allowed[result.TargetSupernodeAccount]; !ok {
+			continue
+		}
+		out = append(out, result)
+	}
+	return out
+}
+
 func storageProofCoverageComplete(results []*audittypes.StorageProofResult, targets []string) (bool, string) {
 	if len(targets) == 0 {
 		return true, ""
@@ -436,7 +474,7 @@ func (s *Service) cascadeKademliaDBBytes(_ context.Context) (uint64, bool) {
 	return total, true
 }
 
-func (s *Service) buildStorageChallengeObservations(ctx context.Context, epochID uint64, requiredOpenPorts []uint32, targets []string) []*audittypes.StorageChallengeObservation {
+func (s *Service) buildStorageChallengeObservations(ctx context.Context, epochID uint64, requiredOpenPorts []uint32, targets []auditmod.AssignedTarget) []*audittypes.StorageChallengeObservation {
 	if len(targets) == 0 {
 		return nil
 	}
@@ -445,7 +483,7 @@ func (s *Service) buildStorageChallengeObservations(ctx context.Context, epochID
 
 	type workItem struct {
 		index  int
-		target string
+		target auditmod.AssignedTarget
 	}
 
 	work := make(chan workItem)
@@ -485,17 +523,19 @@ func (s *Service) buildStorageChallengeObservations(ctx context.Context, epochID
 	return final
 }
 
-func (s *Service) observeTarget(ctx context.Context, epochID uint64, requiredOpenPorts []uint32, target string) *audittypes.StorageChallengeObservation {
-	target = strings.TrimSpace(target)
-	if target == "" {
+func (s *Service) observeTarget(ctx context.Context, epochID uint64, requiredOpenPorts []uint32, target auditmod.AssignedTarget) *audittypes.StorageChallengeObservation {
+	logicalTarget := strings.TrimSpace(target.LogicalAccount)
+	currentTarget := strings.TrimSpace(target.CurrentAccount)
+	if logicalTarget == "" || currentTarget == "" {
 		return nil
 	}
 
-	host, err := s.targetHost(ctx, target)
+	host, err := s.targetHost(ctx, currentTarget)
 	if err != nil {
 		logtrace.Warn(ctx, "storage challenge observe target: resolve host failed", logtrace.Fields{
 			"epoch_id": epochID,
-			"target":   target,
+			"target":   logicalTarget,
+			"current":  currentTarget,
 			"error":    err.Error(),
 		})
 		host = ""
@@ -507,7 +547,7 @@ func (s *Service) observeTarget(ctx context.Context, epochID uint64, requiredOpe
 	}
 
 	return &audittypes.StorageChallengeObservation{
-		TargetSupernodeAccount: target,
+		TargetSupernodeAccount: logicalTarget,
 		PortStates:             portStates,
 	}
 }