diff --git a/docs/services/storage/ttxdb.md b/docs/services/storage/ttxdb.md index a2a4ce596e..bc62a68c16 100644 --- a/docs/services/storage/ttxdb.md +++ b/docs/services/storage/ttxdb.md @@ -88,4 +88,16 @@ The following example shows how to retrieve the total amount of transactions for } fmt.Printf("Transaction: %s\n", tx.ID()) } -``` \ No newline at end of file +``` + +## Composite Owners + +An output owned by a composite identity (multisig, boolpolicy) carries one +audit row per member, so identity consumers (`ByRecipient`, +`RevocationHandles`) see every member. Amount aggregations — payments, +holdings, transaction records — count each `(output index, enrollment ID)` +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`). diff --git a/integration/token/fungible/views/auditor.go b/integration/token/fungible/views/auditor.go index 0e87e3d437..3c2d19e6dc 100644 --- a/integration/token/fungible/views/auditor.go +++ b/integration/token/fungible/views/auditor.go @@ -74,8 +74,8 @@ func (a *AuditView) Call(context view.Context) (any, error) { continue } // compute the payment done in the transaction - sent := inputs.ByEnrollmentID(eID).ByType(tokenType).Sum() - received := outputs.ByEnrollmentID(eID).ByType(tokenType).Sum() + sent := inputs.ByEnrollmentID(eID).ByType(tokenType).UniquePerInput().Sum() + received := outputs.ByEnrollmentID(eID).ByType(tokenType).UniquePerOutput().Sum() fmt.Printf("Payment Limit: [%s] Sent [%d], Received [%d], type [%s]\n", eID, sent.Int64(), received.Int64(), tokenType) diff := big.NewInt(0).Sub(sent, received) @@ -97,8 +97,8 @@ func (a *AuditView) Call(context view.Context) (any, error) { continue } // compute the payment done in the transaction - sent := inputs.ByEnrollmentID(eID).ByType(tokenType).Sum() - received := outputs.ByEnrollmentID(eID).ByType(tokenType).Sum() + sent := inputs.ByEnrollmentID(eID).ByType(tokenType).UniquePerInput().Sum() + received := outputs.ByEnrollmentID(eID).ByType(tokenType).UniquePerOutput().Sum() fmt.Printf("Cumulative Limit: [%s] Sent [%d], Received [%d], type [%s]\n", eID, sent.Int64(), received.Int64(), tokenType) diff := sent.Sub(sent, received) @@ -129,8 +129,8 @@ func (a *AuditView) Call(context view.Context) (any, error) { continue } // compute the amount received - received := outputs.ByEnrollmentID(eID).ByType(tokenType).Sum() - sent := inputs.ByEnrollmentID(eID).ByType(tokenType).Sum() + received := outputs.ByEnrollmentID(eID).ByType(tokenType).UniquePerOutput().Sum() + sent := inputs.ByEnrollmentID(eID).ByType(tokenType).UniquePerInput().Sum() fmt.Printf("Holding Limit: [%s] Sent [%d], Received [%d], type [%s]\n", eID, sent.Int64(), received.Int64(), tokenType) diff := received.Sub(received, sent) diff --git a/integration/token/interop/views/auditor.go b/integration/token/interop/views/auditor.go index 9a54518aa2..9005d66222 100644 --- a/integration/token/interop/views/auditor.go +++ b/integration/token/interop/views/auditor.go @@ -51,8 +51,8 @@ func (a *AuditView) Call(context view.Context) (any, error) { assert.NotEmpty(eID, "enrollment id should not be empty") for _, tokenType := range tokenTypes { // compute the payment done in the transaction - sent := inputs.ByEnrollmentID(eID).ByType(tokenType).Sum() - received := outputs.ByEnrollmentID(eID).ByType(tokenType).Sum() + sent := inputs.ByEnrollmentID(eID).ByType(tokenType).UniquePerInput().Sum() + received := outputs.ByEnrollmentID(eID).ByType(tokenType).UniquePerOutput().Sum() logger.Debugf("Payment Limit: [%s] Sent [%d], Received [%d], type [%s]", eID, sent.Int64(), received.Int64(), tokenType) diff := big.NewInt(0).Sub(sent, received) diff --git a/token/request_composite_output_test.go b/token/request_composite_output_test.go new file mode 100644 index 0000000000..57145e27f4 --- /dev/null +++ b/token/request_composite_output_test.go @@ -0,0 +1,224 @@ +/* +Copyright IBM Corp. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 +*/ + +package token + +import ( + "testing" + + "github.com/LFDT-Panurus/panurus/token/driver" + driver2 "github.com/LFDT-Panurus/panurus/token/driver/mock" + "github.com/LFDT-Panurus/panurus/token/driver/protos-go/v1/request" + "github.com/LFDT-Panurus/panurus/token/services/logging" + "github.com/LFDT-Panurus/panurus/token/token" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// stubLedgerOutput is a minimal driver.Output for extract*Outputs tests. +type stubLedgerOutput struct { + raw []byte +} + +func (s *stubLedgerOutput) Serialize() ([]byte, error) { return s.raw, nil } +func (s *stubLedgerOutput) IsRedeem() bool { return false } +func (s *stubLedgerOutput) GetOwner() []byte { return []byte("policy-owner") } + +// compositeRecipients returns the two member identities and the output +// metadata carrying them as receivers of a single ledger output. +func compositeRecipients() ([]Identity, []*driver.AuditableIdentity) { + m0 := Identity("member-0") + m1 := Identity("member-1") + receivers := []*driver.AuditableIdentity{ + {Identity: m0, AuditInfo: []byte("audit-info-0")}, + {Identity: m1, AuditInfo: []byte("audit-info-1")}, + } + + return []Identity{m0, m1}, receivers +} + +// newCompositeTransferRequest builds a Request holding one transfer action +// with a single 40-unit output owned by a composite identity whose two +// members are enumerated as recipients. +func newCompositeTransferRequest(t *testing.T, ws *driver2.WalletService) *Request { + t.Helper() + recipients, receivers := compositeRecipients() + + transferAction := &driver2.TransferAction{} + transferAction.NumInputsReturns(0) + transferAction.NumOutputsReturns(1) + transferAction.GetOutputsReturns([]driver.Output{&stubLedgerOutput{raw: []byte("ledger-output")}}) + + transferService := &driver2.TransferService{} + transferService.DeserializeTransferActionReturns(transferAction, nil) + + tokensService := &driver2.TokensService{} + tokensService.DeobfuscateReturns( + &token.Token{Owner: Identity("policy-owner"), Type: "USD", Quantity: "0x28"}, + nil, recipients, token.Format("fabtoken"), nil) + + pp := &driver2.PublicParameters{} + pp.PrecisionReturns(64) + ppm := &driver2.PublicParamsManager{} + ppm.PublicParametersReturns(pp) + + tms := &driver2.TokenManagerService{} + tms.TransferServiceReturns(transferService) + tms.IssueServiceReturns(&driver2.IssueService{}) + tms.TokensServiceReturns(tokensService) + tms.WalletServiceReturns(ws) + tms.PublicParamsManagerReturns(ppm) + + return &Request{ + Anchor: "composite-anchor", + Actions: &driver.TokenRequest{ + Actions: []*driver.TypedAction{ + {Type: request.ActionType_ACTION_TYPE_TRANSFER, Raw: []byte("transfer1")}, + }, + }, + Metadata: &driver.TokenRequestMetadata{ + Actions: []*driver.ActionMetadataEntry{ + { + ActionID: 0, + TransferMetadata: &driver.TransferMetadata{ + Outputs: []*driver.TransferOutputMetadata{ + { + OutputMetadata: []byte("output-metadata"), + OutputAuditInfo: []byte("output-audit-info"), + Receivers: receivers, + }, + }, + }, + }, + }, + }, + TokenService: &ManagementService{ + tms: tms, + logger: logging.MustGetLogger(), + }, + } +} + +// TestRequest_Outputs_CompositeSameEIDKeepsMemberRows checks that every +// member of a composite owner keeps its own output row (identity and +// revocation handle stay visible), while UniquePerOutput collapses the rows +// so eid-keyed sums count the output amount exactly once. +func TestRequest_Outputs_CompositeSameEIDKeepsMemberRows(t *testing.T) { + ws := &driver2.WalletService{} + ws.GetEIDAndRHReturnsOnCall(0, "wallet-42", "rh-0", nil) + ws.GetEIDAndRHReturnsOnCall(1, "wallet-42", "rh-1", nil) + + outputs, err := newCompositeTransferRequest(t, ws).Outputs(t.Context()) + require.NoError(t, err) + + // one row per member, same physical output index + require.Equal(t, 2, outputs.Count()) + assert.Equal(t, "wallet-42", outputs.At(0).EnrollmentID) + assert.Equal(t, "wallet-42", outputs.At(1).EnrollmentID) + assert.Equal(t, outputs.At(0).Index, outputs.At(1).Index) + + // identity consumers still see every member + assert.Equal(t, 1, outputs.ByRecipient(Identity("member-0")).Count()) + assert.Equal(t, 1, outputs.ByRecipient(Identity("member-1")).Count()) + assert.Equal(t, []string{"rh-0", "rh-1"}, outputs.RevocationHandles()) + + // economic view counts the amount once + assert.Equal(t, "40", outputs.ByEnrollmentID("wallet-42").UniquePerOutput().Sum().String()) +} + +// TestRequest_Outputs_CompositeDistinctEIDsKeepRows checks that members +// spanning enrollments keep one row each (cross-enrollment visibility). +func TestRequest_Outputs_CompositeDistinctEIDsKeepRows(t *testing.T) { + ws := &driver2.WalletService{} + ws.GetEIDAndRHReturnsOnCall(0, "wallet-42", "", nil) + ws.GetEIDAndRHReturnsOnCall(1, "wallet-43", "", nil) + + outputs, err := newCompositeTransferRequest(t, ws).Outputs(t.Context()) + require.NoError(t, err) + + require.Equal(t, 2, outputs.Count()) + assert.Equal(t, "wallet-42", outputs.At(0).EnrollmentID) + assert.Equal(t, "wallet-43", outputs.At(1).EnrollmentID) +} + +// newCompositeIssueRequest is the issue-action twin of +// newCompositeTransferRequest: one 40-unit issued output, two members. +func newCompositeIssueRequest(t *testing.T, ws *driver2.WalletService) *Request { + t.Helper() + recipients, receivers := compositeRecipients() + issuer := Identity("issuer-1") + + issueAction := &driver2.IssueAction{} + issueAction.NumInputsReturns(0) + issueAction.NumOutputsReturns(1) + issueAction.GetOutputsReturns([]driver.Output{&stubLedgerOutput{raw: []byte("ledger-output")}}) + issueAction.GetIssuerReturns(issuer) + + issueService := &driver2.IssueService{} + issueService.DeserializeIssueActionReturns(issueAction, nil) + + tokensService := &driver2.TokensService{} + tokensService.DeobfuscateReturns( + &token.Token{Owner: Identity("policy-owner"), Type: "USD", Quantity: "0x28"}, + issuer, recipients, token.Format("fabtoken"), nil) + + pp := &driver2.PublicParameters{} + pp.PrecisionReturns(64) + ppm := &driver2.PublicParamsManager{} + ppm.PublicParametersReturns(pp) + + tms := &driver2.TokenManagerService{} + tms.IssueServiceReturns(issueService) + tms.TransferServiceReturns(&driver2.TransferService{}) + tms.TokensServiceReturns(tokensService) + tms.WalletServiceReturns(ws) + tms.PublicParamsManagerReturns(ppm) + + return &Request{ + Anchor: "composite-issue-anchor", + Actions: &driver.TokenRequest{ + Actions: []*driver.TypedAction{ + {Type: request.ActionType_ACTION_TYPE_ISSUE, Raw: []byte("issue1")}, + }, + }, + Metadata: &driver.TokenRequestMetadata{ + Actions: []*driver.ActionMetadataEntry{ + { + ActionID: 0, + IssueMetadata: &driver.IssueMetadata{ + Issuer: driver.AuditableIdentity{Identity: issuer}, + Outputs: []*driver.IssueOutputMetadata{ + { + OutputMetadata: []byte("output-metadata"), + OutputAuditInfo: []byte("output-audit-info"), + Receivers: receivers, + }, + }, + }, + }, + }, + }, + TokenService: &ManagementService{ + tms: tms, + logger: logging.MustGetLogger(), + }, + } +} + +// TestRequest_Outputs_IssueCompositeSameEIDKeepsMemberRows is the issue-side +// twin of TestRequest_Outputs_CompositeSameEIDKeepsMemberRows. +func TestRequest_Outputs_IssueCompositeSameEIDKeepsMemberRows(t *testing.T) { + ws := &driver2.WalletService{} + ws.GetEIDAndRHReturnsOnCall(0, "wallet-42", "rh-0", nil) + ws.GetEIDAndRHReturnsOnCall(1, "wallet-42", "rh-1", nil) + + outputs, err := newCompositeIssueRequest(t, ws).Outputs(t.Context()) + require.NoError(t, err) + + require.Equal(t, 2, outputs.Count()) + assert.Equal(t, []string{"rh-0", "rh-1"}, outputs.RevocationHandles()) + assert.Equal(t, "40", outputs.ByEnrollmentID("wallet-42").UniquePerOutput().Sum().String()) +} diff --git a/token/services/storage/ttxdb/store.go b/token/services/storage/ttxdb/store.go index bc2cc04101..d175657cb6 100644 --- a/token/services/storage/ttxdb/store.go +++ b/token/services/storage/ttxdb/store.go @@ -385,7 +385,7 @@ func TransactionRecords(ctx context.Context, record *token.AuditRecord, timestam outTT := ous.TokenTypes() for _, outEID := range outEIDs { for _, tokenType := range outTT { - received := ous.ByEnrollmentID(outEID).ByType(tokenType).Sum() + received := ous.ByEnrollmentID(outEID).ByType(tokenType).UniquePerOutput().Sum() if received.Cmp(big.NewInt(0)) <= 0 { continue } @@ -431,8 +431,8 @@ func Movements(ctx context.Context, record *token.AuditRecord, created time.Time for _, eID := range eIDs { for _, tokenType := range tokenTypes { - received := outputs.ByEnrollmentID(eID).ByType(tokenType).Sum() - sent := inputs.ByEnrollmentID(eID).ByType(tokenType).Sum() + received := outputs.ByEnrollmentID(eID).ByType(tokenType).UniquePerOutput().Sum() + sent := inputs.ByEnrollmentID(eID).ByType(tokenType).UniquePerInput().Sum() diff := new(big.Int).Sub(received, sent) if sent.Cmp(received) == 0 { continue diff --git a/token/services/storage/ttxdb/store_test.go b/token/services/storage/ttxdb/store_test.go index 118d22eeca..9cf01290bb 100644 --- a/token/services/storage/ttxdb/store_test.go +++ b/token/services/storage/ttxdb/store_test.go @@ -325,3 +325,110 @@ func redeem() token.AuditRecord { Outputs: token.NewOutputStream([]*token.Output{output1}, 64), } } + +// compositePolicySpend models a policy wallet paying 6 to a bank with 34 +// change, where both the spent input and the change output are expanded into +// one row per member of the composite owner: same enrollment ID, same +// physical token. +func compositePolicySpend() token.AuditRecord { + // distinct pointers to the same token ID value, as extraction produces + inputMember0 := &token.Input{ + ActionIndex: 0, + Id: &token2.ID{TxId: "spent-tx", Index: 0}, + EnrollmentID: "policy", + Type: "TOK", + Quantity: token2.NewQuantityFromUInt64(40), + } + inputMember1 := &token.Input{ + ActionIndex: 0, + Id: &token2.ID{TxId: "spent-tx", Index: 0}, + EnrollmentID: "policy", + Type: "TOK", + Quantity: token2.NewQuantityFromUInt64(40), + } + bankOutput := &token.Output{ + ActionIndex: 0, + Index: 1, + EnrollmentID: "bank", + Type: "TOK", + Quantity: token2.NewQuantityFromUInt64(6), + } + changeMember0 := &token.Output{ + ActionIndex: 0, + Index: 2, + EnrollmentID: "policy", + Type: "TOK", + Quantity: token2.NewQuantityFromUInt64(34), + } + changeMember1 := &token.Output{ + ActionIndex: 0, + Index: 2, + EnrollmentID: "policy", + Type: "TOK", + Quantity: token2.NewQuantityFromUInt64(34), + } + + return token.AuditRecord{ + Anchor: "test-composite", + Inputs: token.NewInputStream(qsMock{}, []*token.Input{inputMember0, inputMember1}, 64), + Outputs: token.NewOutputStream([]*token.Output{bankOutput, changeMember0, changeMember1}, 64), + } +} + +// TestMovementRecords_CompositePolicySpend checks the member-expanded change +// rows count once: the payer moves -6 and the bank +6. +func TestMovementRecords_CompositePolicySpend(t *testing.T) { + now := time.Now() + input := compositePolicySpend() + recs, err := ttxdb.Movements(t.Context(), &input, now) + require.NoError(t, err) + assert.Equal(t, []driver.MovementRecord{ + { + TxID: string(input.Anchor), + EnrollmentID: "policy", + TokenType: "TOK", + Amount: big.NewInt(-6), + Timestamp: now, + Status: driver.Pending, + }, + { + TxID: string(input.Anchor), + EnrollmentID: "bank", + TokenType: "TOK", + Amount: big.NewInt(6), + Timestamp: now, + Status: driver.Pending, + }, + }, recs) +} + +// TestTransactionRecords_CompositePolicySpend checks the change amount is +// recorded once despite the per-member rows. +func TestTransactionRecords_CompositePolicySpend(t *testing.T) { + now := time.Now() + input := compositePolicySpend() + recs, err := ttxdb.TransactionRecords(t.Context(), &input, now) + require.NoError(t, err) + assert.Equal(t, []driver.TransactionRecord{ + { + TxID: string(input.Anchor), + ActionType: driver.Transfer, + SenderEID: "policy", + RecipientEID: "bank", + TokenType: "TOK", + Amount: big.NewInt(6), + Timestamp: now, + Status: driver.Pending, + }, + { + TxID: string(input.Anchor), + ActionType: driver.Transfer, + SenderEID: "policy", + RecipientEID: "policy", + TokenType: "TOK", + Amount: big.NewInt(34), + Timestamp: now, + Status: driver.Pending, + }, + }, recs) +} diff --git a/token/stream.go b/token/stream.go index 81d861b1f9..c864353272 100644 --- a/token/stream.go +++ b/token/stream.go @@ -91,6 +91,27 @@ func (o *OutputStream) ByType(typ token.Type) *OutputStream { }) } +// UniquePerOutput returns a stream keeping, for each (Index, EnrollmentID) +// pair, only the first output, so amount aggregation counts a composite +// owner's members once. Identity consumers use the full stream instead. +func (o *OutputStream) UniquePerOutput() *OutputStream { + type key struct { + index uint64 + eID string + } + seen := map[key]bool{} + + return o.Filter(func(t *Output) bool { + k := key{index: t.Index, eID: t.EnrollmentID} + if seen[k] { + return false + } + seen[k] = true + + return true + }) +} + // IsRedeem returns true if this output is a redeem, i.e., it has no owner. func (o *Output) IsRedeem() bool { return len(o.Owner) == 0 @@ -370,6 +391,31 @@ 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. +func (is *InputStream) UniquePerInput() *InputStream { + type key struct { + id token.ID + eID string + } + seen := map[key]bool{} + + return is.Filter(func(t *Input) bool { + if t.Id == nil { + return true + } + k := key{id: *t.Id, eID: t.EnrollmentID} + if seen[k] { + return false + } + seen[k] = true + + return true + }) +} + // Sum returns the sum of the quantities of the inputs. func (is *InputStream) Sum() *big.Int { sum := big.NewInt(0) diff --git a/token/stream_test.go b/token/stream_test.go index 48b36d17b2..6d2e23dcbb 100644 --- a/token/stream_test.go +++ b/token/stream_test.go @@ -252,6 +252,90 @@ func TestOutputStream_ByEnrollmentID(t *testing.T) { assert.Equal(t, []*Output{output1, output3}, filtered.Outputs()) } +func TestOutputStream_UniquePerOutput(t *testing.T) { + cases := []struct { + name string + outputs []*Output + want []*Output + }{ + { + "same index and enrollment ID collapse to the first row", + []*Output{ + {Index: 0, EnrollmentID: "enroll1", RevocationHandler: "first"}, + {Index: 0, EnrollmentID: "enroll1", RevocationHandler: "second"}, + }, + []*Output{{Index: 0, EnrollmentID: "enroll1", RevocationHandler: "first"}}, + }, + { + "same index with different enrollment IDs both survive", + []*Output{{Index: 0, EnrollmentID: "enroll1"}, {Index: 0, EnrollmentID: "enroll2"}}, + []*Output{{Index: 0, EnrollmentID: "enroll1"}, {Index: 0, EnrollmentID: "enroll2"}}, + }, + { + "different indexes with the same enrollment ID both survive", + []*Output{{Index: 0, EnrollmentID: "enroll1"}, {Index: 1, EnrollmentID: "enroll1"}}, + []*Output{{Index: 0, EnrollmentID: "enroll1"}, {Index: 1, EnrollmentID: "enroll1"}}, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + unique := NewOutputStream(tc.outputs, 0).UniquePerOutput() + assert.Equal(t, tc.want, unique.Outputs()) + }) + } +} + +func TestInputStream_UniquePerInput(t *testing.T) { + // distinct pointers to the same token ID value must collapse + cases := []struct { + name string + inputs []*Input + want []*Input + }{ + { + "same token ID and enrollment ID collapse to the first row", + []*Input{ + {Id: &token.ID{TxId: "tx0", Index: 0}, EnrollmentID: "enroll1", RevocationHandler: "first"}, + {Id: &token.ID{TxId: "tx0", Index: 0}, EnrollmentID: "enroll1", RevocationHandler: "second"}, + }, + []*Input{{Id: &token.ID{TxId: "tx0", Index: 0}, EnrollmentID: "enroll1", RevocationHandler: "first"}}, + }, + { + "same token ID with different enrollment IDs both survive", + []*Input{ + {Id: &token.ID{TxId: "tx0", Index: 0}, EnrollmentID: "enroll1"}, + {Id: &token.ID{TxId: "tx0", Index: 0}, EnrollmentID: "enroll2"}, + }, + []*Input{ + {Id: &token.ID{TxId: "tx0", Index: 0}, EnrollmentID: "enroll1"}, + {Id: &token.ID{TxId: "tx0", Index: 0}, EnrollmentID: "enroll2"}, + }, + }, + { + "different token IDs with the same enrollment ID both survive", + []*Input{ + {Id: &token.ID{TxId: "tx0", Index: 0}, EnrollmentID: "enroll1"}, + {Id: &token.ID{TxId: "tx0", Index: 1}, EnrollmentID: "enroll1"}, + }, + []*Input{ + {Id: &token.ID{TxId: "tx0", Index: 0}, EnrollmentID: "enroll1"}, + {Id: &token.ID{TxId: "tx0", Index: 1}, EnrollmentID: "enroll1"}, + }, + }, + { + "inputs with no token ID are all kept", + []*Input{{EnrollmentID: "enroll1"}, {EnrollmentID: "enroll1"}}, + []*Input{{EnrollmentID: "enroll1"}, {EnrollmentID: "enroll1"}}, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + unique := NewInputStream(nil, tc.inputs, 0).UniquePerInput() + assert.Equal(t, tc.want, unique.Inputs()) + }) + } +} + func TestOwnerStream_Count(t *testing.T) { owners := []string{"owner1", "owner2", "owner1"} stream := NewOwnerStream(owners)