From 3bbfaca8c45b2430306114ba0718ab287d4dacea Mon Sep 17 00:00:00 2001 From: Fabio Kenji Date: Fri, 3 Dec 2021 15:40:13 -0800 Subject: [PATCH 01/17] Add more logs --- service/app/contract_service.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/service/app/contract_service.go b/service/app/contract_service.go index cf83116..7825775 100644 --- a/service/app/contract_service.go +++ b/service/app/contract_service.go @@ -910,6 +910,8 @@ func (svc *ContractService) UpdateCirculatingPackContract(ctx context.Context, d "packFlowID": pack.FlowID, }) + eventLogger.Info("handling event...") + switch eventName { // -- REVEAL_REQUEST, Owner has requested to reveal a pack ------------ case REVEAL_REQUEST: @@ -917,6 +919,8 @@ func (svc *ContractService) UpdateCirculatingPackContract(ctx context.Context, d // Make sure the pack is in correct state if err := pack.RevealRequestHandled(); err != nil { err := fmt.Errorf("error while handling %s: %w", eventName, err) + eventLogger.Warn(err.Error()) + eventLogger.Warn(fmt.Sprintf("distID:%s distFlowID:%s packID:%s packFlowID:%s", distribution.ID, distribution.FlowID, pack.ID, packFlow.ID)) return err // rollback } From c18760c03552c67071942d77a3241fea0631174a Mon Sep 17 00:00:00 2001 From: Fabio Kenji Date: Fri, 3 Dec 2021 15:40:37 -0800 Subject: [PATCH 02/17] Fix typo --- service/app/contract_service.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/service/app/contract_service.go b/service/app/contract_service.go index 7825775..2aa15c5 100644 --- a/service/app/contract_service.go +++ b/service/app/contract_service.go @@ -920,7 +920,7 @@ func (svc *ContractService) UpdateCirculatingPackContract(ctx context.Context, d if err := pack.RevealRequestHandled(); err != nil { err := fmt.Errorf("error while handling %s: %w", eventName, err) eventLogger.Warn(err.Error()) - eventLogger.Warn(fmt.Sprintf("distID:%s distFlowID:%s packID:%s packFlowID:%s", distribution.ID, distribution.FlowID, pack.ID, packFlow.ID)) + eventLogger.Warn(fmt.Sprintf("distID:%s distFlowID:%s packID:%s packFlowID:%s", distribution.ID, distribution.FlowID, pack.ID, pack.FlowID)) return err // rollback } From 0a2a08ecc2b1e3692c23d75099b75747b834354d Mon Sep 17 00:00:00 2001 From: Fabio Kenji Date: Fri, 3 Dec 2021 16:31:57 -0800 Subject: [PATCH 03/17] Log info --- service/app/poller.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/service/app/poller.go b/service/app/poller.go index 7654e90..a516de4 100644 --- a/service/app/poller.go +++ b/service/app/poller.go @@ -28,7 +28,7 @@ func poller(app *App) { for { select { case <-ticker.C: - log.Trace("Poll start") + log.Info("Poll start") logPollerRun("handleResolved", handleResolved(ctx, app)) logPollerRun("handleSetup", handleSetup(ctx, app)) @@ -42,7 +42,7 @@ func poller(app *App) { logPollerRun("handleSentTransactions", handleSentTransactions(ctx, app)) logPollerRun("handleSendableTransactions", handleSendableTransactions(ctx, app, transactionRatelimiter)) - log.Trace("Poll end") + log.Info("Poll end") case <-app.quit: cancel() ticker.Stop() From 47325edfaa4e2e9045b45695219c067c9b719019 Mon Sep 17 00:00:00 2001 From: Fabio Kenji Date: Fri, 3 Dec 2021 17:41:10 -0800 Subject: [PATCH 04/17] Ignore state error for now --- service/app/contract_service.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/service/app/contract_service.go b/service/app/contract_service.go index 2aa15c5..cb8ad2e 100644 --- a/service/app/contract_service.go +++ b/service/app/contract_service.go @@ -921,7 +921,8 @@ func (svc *ContractService) UpdateCirculatingPackContract(ctx context.Context, d err := fmt.Errorf("error while handling %s: %w", eventName, err) eventLogger.Warn(err.Error()) eventLogger.Warn(fmt.Sprintf("distID:%s distFlowID:%s packID:%s packFlowID:%s", distribution.ID, distribution.FlowID, pack.ID, pack.FlowID)) - return err // rollback + continue + //return err // rollback } // Update the pack in database From d288bdcb672b541eeb68cc4a8b9407c5165cc378 Mon Sep 17 00:00:00 2001 From: Fabio Kenji Date: Fri, 3 Dec 2021 18:06:13 -0800 Subject: [PATCH 05/17] Log pack info --- service/app/contract_service.go | 1 + 1 file changed, 1 insertion(+) diff --git a/service/app/contract_service.go b/service/app/contract_service.go index cb8ad2e..7703c0b 100644 --- a/service/app/contract_service.go +++ b/service/app/contract_service.go @@ -754,6 +754,7 @@ func (svc *ContractService) UpdateMintingStatus(ctx context.Context, db *gorm.DB // Set the FlowID of the pack // Make sure the pack is in correct state if err := pack.Seal(packFlowID); err != nil { + logger.Warn("pack in wrong state %s packFlowId:%+v", err, packFlowID) return err // rollback } From ae1b4384b63ea337bef52c193b006c0bf3ab84ab Mon Sep 17 00:00:00 2001 From: Fabio Kenji Date: Fri, 3 Dec 2021 18:10:37 -0800 Subject: [PATCH 06/17] FIx log format --- service/app/contract_service.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/service/app/contract_service.go b/service/app/contract_service.go index 7703c0b..f261b45 100644 --- a/service/app/contract_service.go +++ b/service/app/contract_service.go @@ -754,7 +754,7 @@ func (svc *ContractService) UpdateMintingStatus(ctx context.Context, db *gorm.DB // Set the FlowID of the pack // Make sure the pack is in correct state if err := pack.Seal(packFlowID); err != nil { - logger.Warn("pack in wrong state %s packFlowId:%+v", err, packFlowID) + logger.Warn(fmt.Sprintf("pack in wrong state %s packFlowId:%+v", err, packFlowID)) return err // rollback } From 27a277bac293e3817cfb0383c345ac5deb9d2bb1 Mon Sep 17 00:00:00 2001 From: Fabio Kenji Date: Fri, 3 Dec 2021 18:19:00 -0800 Subject: [PATCH 07/17] Continue on wrong state --- service/app/contract_service.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/service/app/contract_service.go b/service/app/contract_service.go index f261b45..b5f0f65 100644 --- a/service/app/contract_service.go +++ b/service/app/contract_service.go @@ -755,7 +755,7 @@ func (svc *ContractService) UpdateMintingStatus(ctx context.Context, db *gorm.DB // Make sure the pack is in correct state if err := pack.Seal(packFlowID); err != nil { logger.Warn(fmt.Sprintf("pack in wrong state %s packFlowId:%+v", err, packFlowID)) - return err // rollback + continue } // Update the pack in database From 2c5c7db80e1c7d8e5ac6b6ab719dcc6b2acc6219 Mon Sep 17 00:00:00 2001 From: Fabio Kenji Date: Tue, 7 Dec 2021 09:09:52 -0800 Subject: [PATCH 08/17] Undo logging changes --- service/app/poller.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/service/app/poller.go b/service/app/poller.go index a516de4..7654e90 100644 --- a/service/app/poller.go +++ b/service/app/poller.go @@ -28,7 +28,7 @@ func poller(app *App) { for { select { case <-ticker.C: - log.Info("Poll start") + log.Trace("Poll start") logPollerRun("handleResolved", handleResolved(ctx, app)) logPollerRun("handleSetup", handleSetup(ctx, app)) @@ -42,7 +42,7 @@ func poller(app *App) { logPollerRun("handleSentTransactions", handleSentTransactions(ctx, app)) logPollerRun("handleSendableTransactions", handleSendableTransactions(ctx, app, transactionRatelimiter)) - log.Info("Poll end") + log.Trace("Poll end") case <-app.quit: cancel() ticker.Stop() From c352d9b1bbf6adb8313eb2ec6df0d095e37931a7 Mon Sep 17 00:00:00 2001 From: Fabio Kenji Date: Wed, 8 Dec 2021 08:45:34 -0800 Subject: [PATCH 09/17] Fix log --- service/app/contract_service.go | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/service/app/contract_service.go b/service/app/contract_service.go index b5f0f65..6786a83 100644 --- a/service/app/contract_service.go +++ b/service/app/contract_service.go @@ -920,8 +920,7 @@ func (svc *ContractService) UpdateCirculatingPackContract(ctx context.Context, d // Make sure the pack is in correct state if err := pack.RevealRequestHandled(); err != nil { err := fmt.Errorf("error while handling %s: %w", eventName, err) - eventLogger.Warn(err.Error()) - eventLogger.Warn(fmt.Sprintf("distID:%s distFlowID:%s packID:%s packFlowID:%s", distribution.ID, distribution.FlowID, pack.ID, pack.FlowID)) + eventLogger.Warn(fmt.Sprintf("distID:%s distFlowID:%s packID:%s packFlowID:%s err:%s", distribution.ID, distribution.FlowID, pack.ID, pack.FlowID, err.Error())) continue //return err // rollback } From b4142eb2972d72dbe8636a4e0f94c5a995f06ddd Mon Sep 17 00:00:00 2001 From: Fabio Kenji Date: Wed, 8 Dec 2021 09:50:52 -0800 Subject: [PATCH 10/17] Just log for now --- service/app/contract_service.go | 2 -- 1 file changed, 2 deletions(-) diff --git a/service/app/contract_service.go b/service/app/contract_service.go index 6786a83..143e919 100644 --- a/service/app/contract_service.go +++ b/service/app/contract_service.go @@ -755,7 +755,6 @@ func (svc *ContractService) UpdateMintingStatus(ctx context.Context, db *gorm.DB // Make sure the pack is in correct state if err := pack.Seal(packFlowID); err != nil { logger.Warn(fmt.Sprintf("pack in wrong state %s packFlowId:%+v", err, packFlowID)) - continue } // Update the pack in database @@ -921,7 +920,6 @@ func (svc *ContractService) UpdateCirculatingPackContract(ctx context.Context, d if err := pack.RevealRequestHandled(); err != nil { err := fmt.Errorf("error while handling %s: %w", eventName, err) eventLogger.Warn(fmt.Sprintf("distID:%s distFlowID:%s packID:%s packFlowID:%s err:%s", distribution.ID, distribution.FlowID, pack.ID, pack.FlowID, err.Error())) - continue //return err // rollback } From e0500ca568ba0811b31625ea7aa67f102b9da145 Mon Sep 17 00:00:00 2001 From: Fabio Kenji Date: Wed, 8 Dec 2021 09:56:21 -0800 Subject: [PATCH 11/17] Just log for now --- service/app/contract_service.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/service/app/contract_service.go b/service/app/contract_service.go index 143e919..a704171 100644 --- a/service/app/contract_service.go +++ b/service/app/contract_service.go @@ -755,6 +755,7 @@ func (svc *ContractService) UpdateMintingStatus(ctx context.Context, db *gorm.DB // Make sure the pack is in correct state if err := pack.Seal(packFlowID); err != nil { logger.Warn(fmt.Sprintf("pack in wrong state %s packFlowId:%+v", err, packFlowID)) + return err //rollback } // Update the pack in database @@ -920,7 +921,7 @@ func (svc *ContractService) UpdateCirculatingPackContract(ctx context.Context, d if err := pack.RevealRequestHandled(); err != nil { err := fmt.Errorf("error while handling %s: %w", eventName, err) eventLogger.Warn(fmt.Sprintf("distID:%s distFlowID:%s packID:%s packFlowID:%s err:%s", distribution.ID, distribution.FlowID, pack.ID, pack.FlowID, err.Error())) - //return err // rollback + return err // rollback } // Update the pack in database From 71f3b354c5950e6bb656e7a5f3258ab2536fbb7b Mon Sep 17 00:00:00 2001 From: Fabio Kenji Date: Wed, 8 Dec 2021 15:05:46 -0800 Subject: [PATCH 12/17] Skip on errors --- service/app/contract_service.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/service/app/contract_service.go b/service/app/contract_service.go index a704171..26d524d 100644 --- a/service/app/contract_service.go +++ b/service/app/contract_service.go @@ -755,7 +755,7 @@ func (svc *ContractService) UpdateMintingStatus(ctx context.Context, db *gorm.DB // Make sure the pack is in correct state if err := pack.Seal(packFlowID); err != nil { logger.Warn(fmt.Sprintf("pack in wrong state %s packFlowId:%+v", err, packFlowID)) - return err //rollback + continue } // Update the pack in database @@ -921,7 +921,7 @@ func (svc *ContractService) UpdateCirculatingPackContract(ctx context.Context, d if err := pack.RevealRequestHandled(); err != nil { err := fmt.Errorf("error while handling %s: %w", eventName, err) eventLogger.Warn(fmt.Sprintf("distID:%s distFlowID:%s packID:%s packFlowID:%s err:%s", distribution.ID, distribution.FlowID, pack.ID, pack.FlowID, err.Error())) - return err // rollback + continue } // Update the pack in database From be41ef3058f142d668594235e3b1a954bb29e3a6 Mon Sep 17 00:00:00 2001 From: Fabio Kenji Date: Wed, 8 Dec 2021 16:00:13 -0800 Subject: [PATCH 13/17] Add more logs --- service/app/contract_service.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/service/app/contract_service.go b/service/app/contract_service.go index 26d524d..7b739fd 100644 --- a/service/app/contract_service.go +++ b/service/app/contract_service.go @@ -1010,6 +1010,7 @@ func (svc *ContractService) UpdateCirculatingPackContract(ctx context.Context, d // Make sure the pack is in correct state if err := pack.Reveal(); err != nil { err := fmt.Errorf("error while handling %s: %w", eventName, err) + eventLogger.Warn(fmt.Sprintf("distID:%s distFlowID:%s packID:%s packFlowID:%s err:%s", distribution.ID, distribution.FlowID, pack.ID, pack.FlowID, err.Error())) return err // rollback } @@ -1024,6 +1025,7 @@ func (svc *ContractService) UpdateCirculatingPackContract(ctx context.Context, d // Make sure the pack is in correct state if err := pack.OpenRequestHandled(); err != nil { err := fmt.Errorf("error while handling %s: %w", eventName, err) + eventLogger.Warn(fmt.Sprintf("distID:%s distFlowID:%s packID:%s packFlowID:%s err:%s", distribution.ID, distribution.FlowID, pack.ID, pack.FlowID, err.Error())) return err // rollback } From c89a6cd3227284bd48bca1de58f9ba19459461af Mon Sep 17 00:00:00 2001 From: Fabio Kenji Date: Wed, 8 Dec 2021 16:01:37 -0800 Subject: [PATCH 14/17] Add more logs --- service/app/contract_service.go | 1 + 1 file changed, 1 insertion(+) diff --git a/service/app/contract_service.go b/service/app/contract_service.go index 7b739fd..43cc5fb 100644 --- a/service/app/contract_service.go +++ b/service/app/contract_service.go @@ -1094,6 +1094,7 @@ func (svc *ContractService) UpdateCirculatingPackContract(ctx context.Context, d // Make sure the pack is in correct state if err := pack.Open(); err != nil { err := fmt.Errorf("error while handling %s: %w", eventName, err) + eventLogger.Warn(fmt.Sprintf("distID:%s distFlowID:%s packID:%s packFlowID:%s err:%s", distribution.ID, distribution.FlowID, pack.ID, pack.FlowID, err.Error())) return err // rollback } From 5707e04b05f1a5f517cde22fe43f506cf2d5df87 Mon Sep 17 00:00:00 2001 From: Fabio Kenji Date: Wed, 8 Dec 2021 16:02:26 -0800 Subject: [PATCH 15/17] Add skip --- service/app/contract_service.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/service/app/contract_service.go b/service/app/contract_service.go index 43cc5fb..d91e2b2 100644 --- a/service/app/contract_service.go +++ b/service/app/contract_service.go @@ -1011,7 +1011,7 @@ func (svc *ContractService) UpdateCirculatingPackContract(ctx context.Context, d if err := pack.Reveal(); err != nil { err := fmt.Errorf("error while handling %s: %w", eventName, err) eventLogger.Warn(fmt.Sprintf("distID:%s distFlowID:%s packID:%s packFlowID:%s err:%s", distribution.ID, distribution.FlowID, pack.ID, pack.FlowID, err.Error())) - return err // rollback + continue } // Update the pack in database @@ -1026,7 +1026,7 @@ func (svc *ContractService) UpdateCirculatingPackContract(ctx context.Context, d if err := pack.OpenRequestHandled(); err != nil { err := fmt.Errorf("error while handling %s: %w", eventName, err) eventLogger.Warn(fmt.Sprintf("distID:%s distFlowID:%s packID:%s packFlowID:%s err:%s", distribution.ID, distribution.FlowID, pack.ID, pack.FlowID, err.Error())) - return err // rollback + continue } // Update the pack in database @@ -1095,7 +1095,7 @@ func (svc *ContractService) UpdateCirculatingPackContract(ctx context.Context, d if err := pack.Open(); err != nil { err := fmt.Errorf("error while handling %s: %w", eventName, err) eventLogger.Warn(fmt.Sprintf("distID:%s distFlowID:%s packID:%s packFlowID:%s err:%s", distribution.ID, distribution.FlowID, pack.ID, pack.FlowID, err.Error())) - return err // rollback + continue } // Update the pack in database From f25e9c363b13a8005b7b4544f00c3fd72589c09e Mon Sep 17 00:00:00 2001 From: Fabio Kenji Date: Fri, 17 Dec 2021 13:54:44 -0800 Subject: [PATCH 16/17] Increase default size for every call --- main.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/main.go b/main.go index d3a7744..6601b08 100644 --- a/main.go +++ b/main.go @@ -85,7 +85,10 @@ func runServer(cfg *config.Config) error { // Flow client // TODO: WithInsecure()? - flowClient, err := client.New(cfg.AccessAPIHost, grpc.WithInsecure()) + maxSize := 1024 * 1024 * 64 + flowClient, err := client.New(cfg.AccessAPIHost, grpc.WithInsecure(), + grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxSize)), + ) if err != nil { return err } From a14e2bc480ea58065bcb8c225789a2ee9d3f5d10 Mon Sep 17 00:00:00 2001 From: Jude Zhu Date: Thu, 6 Jan 2022 13:49:27 -0800 Subject: [PATCH 17/17] split poller --- service/app/app.go | 3 ++ service/app/poller.go | 60 +++++++++++++++++++++++++++++++++++++--- service/config/config.go | 4 +-- test_lib.go | 14 ++-------- 4 files changed, 63 insertions(+), 18 deletions(-) diff --git a/service/app/app.go b/service/app/app.go index cd6d571..78409b9 100644 --- a/service/app/app.go +++ b/service/app/app.go @@ -31,6 +31,9 @@ func New(cfg *config.Config, db *gorm.DB, flowClient *client.Client, poll bool) if poll { go poller(app) + go packContractEventsPoller(app) + go sentTransactionsPoller(app) + go sendableTransactionPoller(app) } return app, nil diff --git a/service/app/poller.go b/service/app/poller.go index 7654e90..187a6fe 100644 --- a/service/app/poller.go +++ b/service/app/poller.go @@ -20,8 +20,6 @@ import ( func poller(app *App) { ticker := time.NewTicker(time.Second) // TODO (latenssi): configurable? - transactionRatelimiter := ratelimit.New(app.cfg.TransactionSendRate) - ctx := context.Background() ctx, cancel := context.WithCancel(ctx) @@ -37,12 +35,66 @@ func poller(app *App) { logPollerRun("handleMinting", handleMinting(ctx, app)) logPollerRun("handleComplete", handleComplete(ctx, app)) + log.Trace("Poll end") + case <-app.quit: + cancel() + ticker.Stop() + return + } + } +} + +// packContractEventsPoller is responsible for checking pack contract events from flow blockchain blocks +func packContractEventsPoller(app *App) { + ticker := time.NewTicker(time.Second) // TODO (latenssi): configurable? + ctx := context.Background() + ctx, cancel := context.WithCancel(ctx) + for { + select { + case <-ticker.C: + log.Trace("PackContractEventsPoller start") logPollerRun("pollCirculatingPackContractEvents", pollCirculatingPackContractEvents(ctx, app)) + log.Trace("PackContractEventsPoller end") + case <-app.quit: + cancel() + ticker.Stop() + return + } + } +} - logPollerRun("handleSentTransactions", handleSentTransactions(ctx, app)) +// sendableTransactionPoller is responsible for checking any sendable to transactions in the transactions table. +func sendableTransactionPoller(app *App) { + ticker := time.NewTicker(time.Second) // TODO (latenssi): configurable? + transactionRatelimiter := ratelimit.New(app.cfg.TransactionSendRate) + + ctx := context.Background() + ctx, cancel := context.WithCancel(ctx) + for { + select { + case <-ticker.C: + log.Trace("SendableTransactionPoller poll start") logPollerRun("handleSendableTransactions", handleSendableTransactions(ctx, app, transactionRatelimiter)) + log.Trace("SendableTransactionPoller poll end") + case <-app.quit: + cancel() + ticker.Stop() + return + } + } +} - log.Trace("Poll end") +// transactionPoller is responsible for sending flow transactions and check transaction status +func sentTransactionsPoller(app *App) { + ticker := time.NewTicker(time.Second) // TODO (latenssi): configurable? + ctx := context.Background() + ctx, cancel := context.WithCancel(ctx) + for { + select { + case <-ticker.C: + log.Trace("SentTransactionsPoller poll start") + logPollerRun("handleSentTransactions", handleSentTransactions(ctx, app)) + log.Trace("SentTransactionsPoller poll end") case <-app.quit: cancel() ticker.Stop() diff --git a/service/config/config.go b/service/config/config.go index 80eaad3..62b1da8 100644 --- a/service/config/config.go +++ b/service/config/config.go @@ -21,8 +21,8 @@ type Config struct { // -- Database -- - DatabaseDSN string `env:"FLOW_PDS_DATABASE_DSN" envDefault:"pds.db"` - DatabaseType string `env:"FLOW_PDS_DATABASE_TYPE" envDefault:"sqlite"` + DatabaseDSN string `env:"FLOW_PDS_DATABASE_DSN" envDefault:"postgresql://pds:pds@localhost:5432/pds"` + DatabaseType string `env:"FLOW_PDS_DATABASE_TYPE" envDefault:"psql"` // -- Host and chain access -- diff --git a/test_lib.go b/test_lib.go index fc694b4..dc1ffe2 100644 --- a/test_lib.go +++ b/test_lib.go @@ -3,7 +3,6 @@ package main import ( "context" "math/big" - "path" "reflect" "strings" "testing" @@ -45,17 +44,8 @@ func getTestCfg(t *testing.T, b *testing.B) *config.Config { panic(err) } - cfg.DatabaseDSN = "test.db" - - if t != nil { - cfg.DatabaseDSN = path.Join(t.TempDir(), "test.db") - } - - if b != nil { - cfg.DatabaseDSN = path.Join(b.TempDir(), "test.db") - } - - cfg.DatabaseType = "sqlite" + cfg.DatabaseDSN = "postgresql://pds:pds@localhost:5432/pds" + cfg.DatabaseType = "psql" return cfg }