fix(token): deduplicate composite-owner inputs whose token IDs are filtered out

UniquePerInput deduplicates by (token ID, enrollment ID) and keeps every
input with a nil token ID. Metadata.filterTransfer strips TokenID from the
input metadata it clones, so consumers of eID-filtered metadata get nil
token IDs and composite-owner member rows are summed once per member again.

Give Input an Index, its position within the action, and fall back to
(action index, input index, enrollment ID) as the deduplication key when
the token ID is absent. Positions survive filtering because filterTransfer
emits one entry per input. Inputs carrying a token ID keep the previous
key, so existing callers are unaffected.

Signed-off-by: Evan <evanyan@sign.global>

Author: EvanYan1024

docs/services/storage/ttxdb.md

@@ -100,4 +100,6 @@ pair once (`OutputStream.UniquePerOutput`), so members sharing an enrollment
 ID do not multiply the recorded amount. A spent input is expanded the same
 way, one row per member carrying the token's full quantity; sent-amount
 aggregation counts each `(token ID, enrollment ID)` pair once
-(`InputStream.UniquePerInput`).
+(`InputStream.UniquePerInput`), falling back to
+`(action index, input index, enrollment ID)` when the token ID has been
+filtered out of the metadata.

token/request.go

@@ -907,9 +907,10 @@ func (r *Request) inputs(ctx context.Context, failOnMissing bool) (*InputStream,
 
 func (r *Request) extractIssueInputs(actionIndex int, metadata *IssueMetadata) ([]*Input, error) {
 	var inputs []*Input
-	for _, input := range metadata.Inputs {
+	for j, input := range metadata.Inputs {
 		inputs = append(inputs, &Input{
 			ActionIndex: actionIndex,
+			Index:       j,
 			Id:          input.TokenID,
 		})
 	}
@@ -940,6 +941,7 @@ func (r *Request) extractTransferInputs(ctx context.Context, actionIndex int, me
 
 			inputs = append(inputs, &Input{
 				ActionIndex:       actionIndex,
+				Index:             j,
 				Id:                metadata.TokenIDAt(j),
 				Owner:             sender.Identity,
 				OwnerAuditInfo:    sender.AuditInfo,

token/stream.go

@@ -232,7 +232,9 @@ func (o *OutputStream) String() string {
 
 // Input models an input of a token action
 type Input struct {
-	ActionIndex       int
+	ActionIndex int
+	// Index is the position of this input within its action
+	Index             int
 	Id                *token.ID
 	Owner             Identity
 	OwnerAuditInfo    []byte
@@ -391,26 +393,40 @@ func (is *InputStream) ByType(tokenType token.Type) *InputStream {
 	})
 }
 
-// UniquePerInput returns a stream keeping, for each (token ID, EnrollmentID)
-// pair, only the first input, so amount aggregation counts a composite
-// owner's members once. Inputs with no token ID are all kept. Identity
-// consumers use the full stream instead.
+// UniquePerInput returns a stream keeping, for each spent token and
+// EnrollmentID, only the first input, so amount aggregation counts a
+// composite owner's members once. The token is identified by its token ID,
+// or by (ActionIndex, Index) when the token ID has been filtered out of the
+// metadata; the two key spaces never mix, since only transfer inputs lose
+// their token ID. Identity consumers use the full stream instead.
 func (is *InputStream) UniquePerInput() *InputStream {
-	type key struct {
+	type idKey struct {
 		id  token.ID
 		eID string
 	}
-	seen := map[key]bool{}
+	type posKey struct {
+		actionIndex int
+		index       int
+		eID         string
+	}
+	seenByID := map[idKey]bool{}
+	seenByPos := map[posKey]bool{}
 
 	return is.Filter(func(t *Input) bool {
-		if t.Id == nil {
+		if t.Id != nil {
+			k := idKey{id: *t.Id, eID: t.EnrollmentID}
+			if seenByID[k] {
+				return false
+			}
+			seenByID[k] = true
+
 			return true
 		}
-		k := key{id: *t.Id, eID: t.EnrollmentID}
-		if seen[k] {
+		k := posKey{actionIndex: t.ActionIndex, index: t.Index, eID: t.EnrollmentID}
+		if seenByPos[k] {
 			return false
 		}
-		seen[k] = true
+		seenByPos[k] = true
 
 		return true
 	})

token/stream_test.go

@@ -286,7 +286,6 @@ func TestOutputStream_UniquePerOutput(t *testing.T) {
 }
 
 func TestInputStream_UniquePerInput(t *testing.T) {
-	// distinct pointers to the same token ID value must collapse
 	cases := []struct {
 		name   string
 		inputs []*Input
@@ -312,7 +311,7 @@ func TestInputStream_UniquePerInput(t *testing.T) {
 			},
 		},
 		{
-			"different token IDs with the same enrollment ID both survive",
+			"different token IDs with the same enrollment ID both survive despite equal positions",
 			[]*Input{
 				{Id: &token.ID{TxId: "tx0", Index: 0}, EnrollmentID: "enroll1"},
 				{Id: &token.ID{TxId: "tx0", Index: 1}, EnrollmentID: "enroll1"},
@@ -323,9 +322,38 @@ func TestInputStream_UniquePerInput(t *testing.T) {
 			},
 		},
 		{
-			"inputs with no token ID are all kept",
-			[]*Input{{EnrollmentID: "enroll1"}, {EnrollmentID: "enroll1"}},
-			[]*Input{{EnrollmentID: "enroll1"}, {EnrollmentID: "enroll1"}},
+			"member rows with a filtered-out token ID collapse on action and input index",
+			[]*Input{
+				{ActionIndex: 0, Index: 0, EnrollmentID: "enroll1", RevocationHandler: "first"},
+				{ActionIndex: 0, Index: 0, EnrollmentID: "enroll1", RevocationHandler: "second"},
+			},
+			[]*Input{{ActionIndex: 0, Index: 0, EnrollmentID: "enroll1", RevocationHandler: "first"}},
+		},
+		{
+			"filtered rows with different positions or enrollment IDs all survive",
+			[]*Input{
+				{ActionIndex: 0, Index: 0, EnrollmentID: "enroll1"},
+				{ActionIndex: 0, Index: 1, EnrollmentID: "enroll1"},
+				{ActionIndex: 1, Index: 0, EnrollmentID: "enroll1"},
+				{ActionIndex: 0, Index: 0, EnrollmentID: "enroll2"},
+			},
+			[]*Input{
+				{ActionIndex: 0, Index: 0, EnrollmentID: "enroll1"},
+				{ActionIndex: 0, Index: 1, EnrollmentID: "enroll1"},
+				{ActionIndex: 1, Index: 0, EnrollmentID: "enroll1"},
+				{ActionIndex: 0, Index: 0, EnrollmentID: "enroll2"},
+			},
+		},
+		{
+			"a row with a token ID never collides with a positionally identical row without one",
+			[]*Input{
+				{ActionIndex: 0, Index: 0, Id: &token.ID{TxId: "tx0", Index: 0}, EnrollmentID: "enroll1"},
+				{ActionIndex: 0, Index: 0, EnrollmentID: "enroll1"},
+			},
+			[]*Input{
+				{ActionIndex: 0, Index: 0, Id: &token.ID{TxId: "tx0", Index: 0}, EnrollmentID: "enroll1"},
+				{ActionIndex: 0, Index: 0, EnrollmentID: "enroll1"},
+			},
 		},
 	}
 	for _, tc := range cases {