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

docs/services/storage/ttxdb.md

@@ -100,4 +100,8 @@ 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. Positional deduplication applies only to inputs
+built through `token.NewInput`, which marks the position as known; a
+literal-built input without a token ID keeps the previous keep-all behavior.

token/request.go

@@ -907,11 +907,8 @@ 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 {
-		inputs = append(inputs, &Input{
-			ActionIndex: actionIndex,
-			Id:          input.TokenID,
-		})
+	for j, input := range metadata.Inputs {
+		inputs = append(inputs, NewInput(actionIndex, j, input.TokenID))
 	}
 
 	return inputs, nil
@@ -938,14 +935,12 @@ func (r *Request) extractTransferInputs(ctx context.Context, actionIndex int, me
 				return nil, errors.Wrapf(err, "failed getting enrollment id and revocation handle [%d,%d]", actionIndex, j)
 			}
 
-			inputs = append(inputs, &Input{
-				ActionIndex:       actionIndex,
-				Id:                metadata.TokenIDAt(j),
-				Owner:             sender.Identity,
-				OwnerAuditInfo:    sender.AuditInfo,
-				EnrollmentID:      eID,
-				RevocationHandler: rID,
-			})
+			in := NewInput(actionIndex, j, metadata.TokenIDAt(j))
+			in.Owner = sender.Identity
+			in.OwnerAuditInfo = sender.AuditInfo
+			in.EnrollmentID = eID
+			in.RevocationHandler = rID
+			inputs = append(inputs, in)
 		}
 	}
 

token/stream.go

@@ -232,14 +232,30 @@ 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; it takes part in
+	// deduplication only on inputs built through NewInput
+	Index             int
 	Id                *token.ID
 	Owner             Identity
 	OwnerAuditInfo    []byte
 	EnrollmentID      string
 	RevocationHandler string
 	Type              token.Type
 	Quantity          token.Quantity
+
+	// positionKnown records that (ActionIndex, Index) was set deliberately —
+	// NewInput stamps it — so UniquePerInput never collapses literal-built
+	// inputs on their zero-valued position
+	positionKnown bool
+}
+
+// NewInput returns an Input positioned within its action. The position stands
+// in for the token ID during deduplication when the metadata has the token ID
+// filtered out (see UniquePerInput), so inputs are built through here; an
+// input built as a literal keeps no position and never collapses on it.
+func NewInput(actionIndex, index int, id *token.ID) *Input {
+	return &Input{ActionIndex: actionIndex, Index: index, Id: id, positionKnown: true}
 }
 
 // InputStream models a stream over a set of inputs (Input).
@@ -391,26 +407,46 @@ 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. The position counts only for inputs built through NewInput:
+// a literal-built input without a token ID is kept as it stands, never
+// collapsed on a position nobody set. 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] {
+		if !t.positionKnown {
+			return true
+		}
+		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

@@ -285,8 +285,17 @@ func TestOutputStream_UniquePerOutput(t *testing.T) {
 	}
 }
 
+// extractedInput builds an input the way the request's extraction path does,
+// so its position takes part in deduplication.
+func extractedInput(actionIndex, index int, id *token.ID, eID, rh string) *Input {
+	in := NewInput(actionIndex, index, id)
+	in.EnrollmentID = eID
+	in.RevocationHandler = rh
+
+	return in
+}
+
 func TestInputStream_UniquePerInput(t *testing.T) {
-	// distinct pointers to the same token ID value must collapse
 	cases := []struct {
 		name   string
 		inputs []*Input
@@ -312,7 +321,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 +332,49 @@ 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{
+				extractedInput(0, 0, nil, "enroll1", "first"),
+				extractedInput(0, 0, nil, "enroll1", "second"),
+			},
+			[]*Input{extractedInput(0, 0, nil, "enroll1", "first")},
+		},
+		{
+			"filtered rows with different positions or enrollment IDs all survive",
+			[]*Input{
+				extractedInput(0, 0, nil, "enroll1", ""),
+				extractedInput(0, 1, nil, "enroll1", ""),
+				extractedInput(1, 0, nil, "enroll1", ""),
+				extractedInput(0, 0, nil, "enroll2", ""),
+			},
+			[]*Input{
+				extractedInput(0, 0, nil, "enroll1", ""),
+				extractedInput(0, 1, nil, "enroll1", ""),
+				extractedInput(1, 0, nil, "enroll1", ""),
+				extractedInput(0, 0, nil, "enroll2", ""),
+			},
+		},
+		{
+			"a row with a token ID never collides with a positionally identical row without one",
+			[]*Input{
+				extractedInput(0, 0, &token.ID{TxId: "tx0", Index: 0}, "enroll1", ""),
+				extractedInput(0, 0, nil, "enroll1", ""),
+			},
+			[]*Input{
+				extractedInput(0, 0, &token.ID{TxId: "tx0", Index: 0}, "enroll1", ""),
+				extractedInput(0, 0, nil, "enroll1", ""),
+			},
+		},
+		{
+			"literal-built rows with no token ID never collapse on the zero-valued position",
+			[]*Input{
+				{EnrollmentID: "enroll1"},
+				{EnrollmentID: "enroll1"},
+			},
+			[]*Input{
+				{EnrollmentID: "enroll1"},
+				{EnrollmentID: "enroll1"},
+			},
 		},
 	}
 	for _, tc := range cases {