Skip to content

Remove redundant storage + relevent code from exchange message tracker #1601

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
vkuznecovas merged 1 commit into from
Jan 31, 2020
Merged

Remove redundant storage + relevent code from exchange message tracker #1601

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 0 additions & 3 deletions cmd/di.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -162,7 +162,6 @@ type Dependencies struct {
Reporter *feedback.Reporter

ProviderInvoiceStorage *pingpong.ProviderInvoiceStorage
ConsumerInvoiceStorage *pingpong.ConsumerInvoiceStorage
ConsumerTotalsStorage *pingpong.ConsumerTotalsStorage
AccountantPromiseStorage *pingpong.AccountantPromiseStorage
ConsumerBalanceTracker *pingpong.ConsumerBalanceTracker
Expand Down Expand Up @@ -387,7 +386,6 @@ func (di *Dependencies) bootstrapStorage(path string) error {

invoiceStorage := pingpong.NewInvoiceStorage(di.Storage)
di.ProviderInvoiceStorage = pingpong.NewProviderInvoiceStorage(invoiceStorage)
di.ConsumerInvoiceStorage = pingpong.NewConsumerInvoiceStorage(invoiceStorage)
di.ConsumerTotalsStorage = pingpong.NewConsumerTotalsStorage(di.Storage)
di.AccountantPromiseStorage = pingpong.NewAccountantPromiseStorage(di.Storage)
return nil
Expand Down Expand Up @@ -512,7 +510,6 @@ func (di *Dependencies) bootstrapNodeComponents(nodeOptions node.Options, tequil
di.Keystore,
nodeOptions,
di.SignerFactory,
di.ConsumerInvoiceStorage,
di.ConsumerTotalsStorage,
nodeOptions.Transactor.ChannelImplementation,
nodeOptions.Transactor.RegistryAddress,
Expand Down
35 changes: 9 additions & 26 deletions session/pingpong/exchange_message_tracker.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,14 +72,14 @@ type ExchangeMessageTracker struct {
channelAddress identity.Identity
receivedFirst bool

deps ExchangeMessageTrackerDeps
lastInvoice crypto.Invoice
deps ExchangeMessageTrackerDeps
}

// ExchangeMessageTrackerDeps contains all the dependencies for the exchange message tracker.
type ExchangeMessageTrackerDeps struct {
InvoiceChan chan crypto.Invoice
PeerExchangeMessageSender PeerExchangeMessageSender
ConsumerInvoiceStorage consumerInvoiceStorage
ConsumerTotalsStorage consumerTotalsStorage
TimeTracker timeTracker
Ks *keystore.KeyStore
Expand All @@ -92,8 +92,9 @@ type ExchangeMessageTrackerDeps struct {
// NewExchangeMessageTracker returns a new instance of exchange message tracker.
func NewExchangeMessageTracker(emtd ExchangeMessageTrackerDeps) *ExchangeMessageTracker {
return &ExchangeMessageTracker{
stop: make(chan struct{}),
deps: emtd,
stop: make(chan struct{}),
deps: emtd,
lastInvoice: crypto.Invoice{},
}
}

Expand Down Expand Up @@ -127,11 +128,7 @@ func (emt *ExchangeMessageTracker) Start() error {
return err
}

err = emt.deps.ConsumerInvoiceStorage.Store(emt.deps.Identity, emt.deps.Peer, invoice)
if err != nil {
return errors.Wrap(err, "could not store invoice")
}

emt.lastInvoice = invoice
}
}
}
Expand Down Expand Up @@ -197,24 +194,14 @@ func (emt *ExchangeMessageTracker) isInvoiceOK(invoice crypto.Invoice) error {
}

func (emt *ExchangeMessageTracker) calculateAmountToPromise(invoice crypto.Invoice) (toPromise uint64, diff uint64, err error) {
previous, err := emt.deps.ConsumerInvoiceStorage.Get(emt.deps.Identity, emt.deps.Peer)
if err != nil {
if err == ErrNotFound {
// do nothing, really
log.Debug().Msg("No previous invoice found, assuming zero")
} else {
return 0, 0, errors.Wrap(err, fmt.Sprintf("could not get previous total for peer %q", invoice.Provider))
}
}

diff = invoice.AgreementTotal - previous.AgreementTotal
diff = invoice.AgreementTotal - emt.lastInvoice.AgreementTotal
totalPromised, err := emt.getGrandTotalPromised()
if err != nil {
return 0, 0, err
}

// This is a new agreement, we need to take in the agreement total and just add it to total promised
if previous.AgreementID != invoice.AgreementID {
if emt.lastInvoice.AgreementID != invoice.AgreementID {
diff = invoice.AgreementTotal
}

Expand Down Expand Up @@ -247,11 +234,7 @@ func (emt *ExchangeMessageTracker) issueExchangeMessage(invoice crypto.Invoice)

// TODO: we'd probably want to check if we have enough balance here
err = emt.incrementGrandTotalPromised(diff)
if err != nil {
return errors.Wrap(err, "could not increment grand total")
}

return emt.deps.ConsumerTotalsStorage.Store(emt.deps.Peer.Address, invoice.AgreementTotal)
return errors.Wrap(err, "could not increment grand total")
}

// Stop stops the message tracker.
Expand Down
85 changes: 19 additions & 66 deletions session/pingpong/exchange_message_tracker_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,12 +65,10 @@ func Test_ExchangeMessageTracker_Start_Stop(t *testing.T) {
defer bolt.Close()

tracker := session.NewTracker(time.Now)
invoiceStorage := NewConsumerInvoiceStorage(NewInvoiceStorage(bolt))
totalsStorage := NewConsumerTotalsStorage(bolt)
deps := ExchangeMessageTrackerDeps{
InvoiceChan: invoiceChan,
PeerExchangeMessageSender: mockSender,
ConsumerInvoiceStorage: invoiceStorage,
ConsumerTotalsStorage: totalsStorage,
TimeTracker: &tracker,
Ks: ks,
Expand Down Expand Up @@ -113,12 +111,10 @@ func Test_ExchangeMessageTracker_SendsMessage(t *testing.T) {
defer bolt.Close()

tracker := session.NewTracker(time.Now)
invoiceStorage := NewConsumerInvoiceStorage(NewInvoiceStorage(bolt))
totalsStorage := NewConsumerTotalsStorage(bolt)
deps := ExchangeMessageTrackerDeps{
InvoiceChan: invoiceChan,
PeerExchangeMessageSender: mockSender,
ConsumerInvoiceStorage: invoiceStorage,
ConsumerTotalsStorage: totalsStorage,
TimeTracker: &tracker,
Publisher: &mockPublisher{},
Expand Down Expand Up @@ -181,12 +177,10 @@ func Test_ExchangeMessageTracker_SendsMessage_OnFreeService(t *testing.T) {
defer bolt.Close()

tracker := session.NewTracker(time.Now)
invoiceStorage := NewConsumerInvoiceStorage(NewInvoiceStorage(bolt))
totalsStorage := NewConsumerTotalsStorage(bolt)
deps := ExchangeMessageTrackerDeps{
InvoiceChan: invoiceChan,
PeerExchangeMessageSender: mockSender,
ConsumerInvoiceStorage: invoiceStorage,
ConsumerTotalsStorage: totalsStorage,
TimeTracker: &tracker,
Publisher: &mockPublisher{},
Expand Down Expand Up @@ -245,13 +239,11 @@ func Test_ExchangeMessageTracker_BubblesErrors(t *testing.T) {
defer bolt.Close()

tracker := session.NewTracker(time.Now)
invoiceStorage := NewConsumerInvoiceStorage(NewInvoiceStorage(bolt))
totalsStorage := NewConsumerTotalsStorage(bolt)
deps := ExchangeMessageTrackerDeps{
InvoiceChan: invoiceChan,
Publisher: &mockPublisher{},
PeerExchangeMessageSender: mockSender,
ConsumerInvoiceStorage: invoiceStorage,
ConsumerTotalsStorage: totalsStorage,
TimeTracker: &tracker,
Ks: ks,
Expand Down Expand Up @@ -479,9 +471,9 @@ func TestExchangeMessageTracker_incrementGrandTotalPromised(t *testing.T) {

func TestExchangeMessageTracker_calculateAmountToPromise(t *testing.T) {
type fields struct {
peer identity.Identity
consumerInvoiceStorage *mockConsumerInvoiceStorage
consumerTotalsStorage *mockConsumerTotalsStorage
peer identity.Identity
lastInvoice crypto.Invoice
consumerTotalsStorage *mockConsumerTotalsStorage
}
tests := []struct {
name string
Expand All @@ -491,20 +483,9 @@ func TestExchangeMessageTracker_calculateAmountToPromise(t *testing.T) {
wantDiff uint64
wantErr bool
}{
{
name: "bubbles invoice storage errors",
fields: fields{
consumerInvoiceStorage: &mockConsumerInvoiceStorage{
err: errors.New("explosions everywhere"),
},
},
invoice: crypto.Invoice{},
wantErr: true,
},
{
name: "bubbles totals storage errors",
fields: fields{
consumerInvoiceStorage: &mockConsumerInvoiceStorage{},
consumerTotalsStorage: &mockConsumerTotalsStorage{
err: errors.New("explosions everywhere"),
},
Expand All @@ -513,11 +494,8 @@ func TestExchangeMessageTracker_calculateAmountToPromise(t *testing.T) {
wantErr: true,
},
{
name: "ignores bolt not found errors",
name: "assumes zero",
fields: fields{
consumerInvoiceStorage: &mockConsumerInvoiceStorage{
err: ErrNotFound,
},
consumerTotalsStorage: &mockConsumerTotalsStorage{},
},
invoice: crypto.Invoice{AgreementTotal: 10},
Expand All @@ -528,9 +506,6 @@ func TestExchangeMessageTracker_calculateAmountToPromise(t *testing.T) {
{
name: "calculates correctly with different grand total",
fields: fields{
consumerInvoiceStorage: &mockConsumerInvoiceStorage{
err: ErrNotFound,
},
consumerTotalsStorage: &mockConsumerTotalsStorage{
res: 100,
},
Expand All @@ -543,9 +518,7 @@ func TestExchangeMessageTracker_calculateAmountToPromise(t *testing.T) {
{
name: "calculates correctly with previous invoice",
fields: fields{
consumerInvoiceStorage: &mockConsumerInvoiceStorage{
res: crypto.Invoice{AgreementID: 111, AgreementTotal: 111},
},
lastInvoice: crypto.Invoice{AgreementID: 111, AgreementTotal: 111},
consumerTotalsStorage: &mockConsumerTotalsStorage{
res: 100,
},
Expand All @@ -560,11 +533,11 @@ func TestExchangeMessageTracker_calculateAmountToPromise(t *testing.T) {
t.Run(tt.name, func(t *testing.T) {
emt := &ExchangeMessageTracker{
deps: ExchangeMessageTrackerDeps{
ConsumerTotalsStorage: tt.fields.consumerTotalsStorage,
Peer: tt.fields.peer,
ConsumerInvoiceStorage: tt.fields.consumerInvoiceStorage,
ConsumerTotalsStorage: tt.fields.consumerTotalsStorage,
Peer: tt.fields.peer,
},
}
emt.lastInvoice = tt.fields.lastInvoice
gotToPromise, gotDiff, err := emt.calculateAmountToPromise(tt.invoice)
if (err != nil) != tt.wantErr {
t.Errorf("ExchangeMessageTracker.calculateAmountToPromise() error = %v, wantErr %v", err, tt.wantErr)
Expand Down Expand Up @@ -604,16 +577,14 @@ func TestExchangeMessageTracker_issueExchangeMessage_publishesEvents(t *testing.
},
ConsumerTotalsStorage: &mockConsumerTotalsStorage{},
Peer: peerID,
ConsumerInvoiceStorage: &mockConsumerInvoiceStorage{
res: crypto.Invoice{
AgreementTotal: 10,
},
},
Ks: ks,
Identity: identity.FromAddress(acc.Address.Hex()),
Publisher: mp,
Ks: ks,
Identity: identity.FromAddress(acc.Address.Hex()),
Publisher: mp,
},
}
emt.lastInvoice = crypto.Invoice{
AgreementTotal: 10,
}
err = emt.issueExchangeMessage(crypto.Invoice{
AgreementTotal: 15,
Hashlock: "0x441Da57A51e42DAB7Daf55909Af93A9b00eEF23C",
Expand Down Expand Up @@ -646,7 +617,7 @@ func TestExchangeMessageTracker_issueExchangeMessage(t *testing.T) {
keystore *keystore.KeyStore
identity identity.Identity
peer identity.Identity
consumerInvoiceStorage *mockConsumerInvoiceStorage
lastInvoice crypto.Invoice
consumerTotalsStorage *mockConsumerTotalsStorage
}
type args struct {
Expand All @@ -659,21 +630,6 @@ func TestExchangeMessageTracker_issueExchangeMessage(t *testing.T) {
wantErr bool
wantMsg *crypto.ExchangeMessage
}{
{
name: "bubbles calculation errors",
fields: fields{
identity: identity.FromAddress(acc.Address.Hex()),
peer: peerID,
keystore: ks,
peerExchangeMessageSender: &MockPeerExchangeMessageSender{
chanToWriteTo: make(chan crypto.ExchangeMessage, 10),
},
consumerInvoiceStorage: &mockConsumerInvoiceStorage{
err: errors.New("explosions everywhere"),
},
},
wantErr: true,
},
{
name: "bubbles exchange message creation errors",
fields: fields{
Expand All @@ -683,8 +639,7 @@ func TestExchangeMessageTracker_issueExchangeMessage(t *testing.T) {
peerExchangeMessageSender: &MockPeerExchangeMessageSender{
chanToWriteTo: make(chan crypto.ExchangeMessage, 10),
},
consumerInvoiceStorage: &mockConsumerInvoiceStorage{},
consumerTotalsStorage: &mockConsumerTotalsStorage{},
consumerTotalsStorage: &mockConsumerTotalsStorage{},
},
wantErr: true,
},
Expand All @@ -698,8 +653,7 @@ func TestExchangeMessageTracker_issueExchangeMessage(t *testing.T) {
chanToWriteTo: make(chan crypto.ExchangeMessage, 10),
mockError: errors.New("explosions everywhere"),
},
consumerInvoiceStorage: &mockConsumerInvoiceStorage{},
consumerTotalsStorage: &mockConsumerTotalsStorage{},
consumerTotalsStorage: &mockConsumerTotalsStorage{},
},
wantErr: false,
},
Expand All @@ -712,8 +666,7 @@ func TestExchangeMessageTracker_issueExchangeMessage(t *testing.T) {
peerExchangeMessageSender: &MockPeerExchangeMessageSender{
chanToWriteTo: make(chan crypto.ExchangeMessage, 10),
},
consumerInvoiceStorage: &mockConsumerInvoiceStorage{},
consumerTotalsStorage: &mockConsumerTotalsStorage{},
consumerTotalsStorage: &mockConsumerTotalsStorage{},
},
args: args{
invoice: crypto.Invoice{
Expand All @@ -731,12 +684,12 @@ func TestExchangeMessageTracker_issueExchangeMessage(t *testing.T) {
PeerExchangeMessageSender: tt.fields.peerExchangeMessageSender,
ConsumerTotalsStorage: tt.fields.consumerTotalsStorage,
Peer: tt.fields.peer,
ConsumerInvoiceStorage: tt.fields.consumerInvoiceStorage,
Ks: tt.fields.keystore,
Identity: tt.fields.identity,
Publisher: &mockPublisher{},
},
}
emt.lastInvoice = tt.fields.lastInvoice
if err := emt.issueExchangeMessage(tt.args.invoice); (err != nil) != tt.wantErr {
t.Errorf("ExchangeMessageTracker.issueExchangeMessage() error = %v, wantErr %v", err, tt.wantErr)
}
Expand Down
2 changes: 0 additions & 2 deletions session/pingpong/factory.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,7 +109,6 @@ func BackwardsCompatibleExchangeFactoryFunc(
keystore *keystore.KeyStore,
options node.Options,
signer identity.SignerFactory,
invoiceStorage consumerInvoiceStorage,
totalStorage consumerTotalsStorage,
channelImplementation string,
registryAddress string, publisher eventbus.Publisher) func(paymentInfo *promise.PaymentInfo,
Expand Down Expand Up @@ -153,7 +152,6 @@ func BackwardsCompatibleExchangeFactoryFunc(
deps := ExchangeMessageTrackerDeps{
InvoiceChan: invoices,
PeerExchangeMessageSender: NewExchangeSender(dialog),
ConsumerInvoiceStorage: invoiceStorage,
ConsumerTotalsStorage: totalStorage,
TimeTracker: &timeTracker,
Ks: keystore,
Expand Down
Loading