[Access] Properly handle subscription errors in data providers #7046

illia-malachyn · 2025-02-17T14:22:54Z

Distinguish between context.Canceled errors originating from the streamer and those triggered by the DataProvider’s Close() method.

I Added closeChan that is used in DataProvider.Close() method to indicate DataProvider.Run() that users of data providers (WebSocket controller in our case) want to finish receiving data.
HandleSubscription() function is replaced by run() function that is aware of closeChan. I made a new function for it because HandleSubscription() is widely used in the access package (HandleRPCSubscription has 22 usages atm).
run() returns nil if closeChan was closed. HandleSubscription returned ctx.Canceled which lead to confusion as ctx.Canceled could come from 2 sources (streamer and websocket controller).
I did a little refactoring to a bunch of sendResponse() functions to make it more readable.

Closes #7040 #7047

Distinguish between `context.Canceled` errors originating from the streamer and those triggered by the DataProvider’s `Close()` method. Use `wasClosedByClient()` to suppress expected cancellations while propagating unexpected ones

codecov-commenter · 2025-02-17T14:26:34Z

Codecov Report

Attention: Patch coverage is 87.50000% with 18 lines in your changes missing coverage. Please review.

Project coverage is 41.24%. Comparing base (bd39edc) to head (feff7e5).

Files with missing lines	Patch %	Lines
...ss/rest/websockets/data_providers/base_provider.go	54.16%	8 Missing and 3 partials ⚠️
...ockets/data_providers/account_statuses_provider.go	90.32%	2 Missing and 1 partial ⚠️
.../rest/websockets/data_providers/events_provider.go	88.46%	2 Missing and 1 partial ⚠️
.../rest/websockets/data_providers/blocks_provider.go	88.88%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #7046      +/-   ##
==========================================
+ Coverage   41.23%   41.24%   +0.01%     
==========================================
  Files        2138     2138              
  Lines      188585   188622      +37     
==========================================
+ Hits        77764    77799      +35     
- Misses     104327   104329       +2     
  Partials     6494     6494

Flag	Coverage Δ
unittests	`41.24% <87.50%> (+0.01%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

Guitarheroua

Looks good to me!

illia-malachyn · 2025-02-17T15:02:49Z

After discussing with Yurii, we agreed to use a different approach here.

As this will changed, will re-aprove final implementation

We use it to distinguish place where cxt.Canceled error comes from. Also, I refactored each data provider's Run() function. Now it's more readable and clear.

peterargue

why is this refactor needed? can you point out the important changes, it's not clear to me

illia-malachyn · 2025-02-20T12:44:28Z

why is this refactor needed? can you point out the important changes, it's not clear to me

Hey. I updated PR's description. Added a context of what has been done and why.

illia-malachyn · 2025-02-20T12:45:29Z

engine/access/rest/websockets/data_providers/account_statuses_provider.go

+	// The block counter increments until either:
+	// 1. The account emits events
+	// 2. The heartbeat interval is reached
+	*blocksSinceLastMessage += 1


all places with blockSinceLastMessage were refactored

Your refactor variant should work as expected, I would just add to comment that:

// Only send a response if there's meaningful data to send or the heartbeat interval limit is reached

illia-malachyn · 2025-02-20T12:46:10Z

engine/access/rest/websockets/data_providers/account_statuses_provider.go

+	}
+
+	var accountStatusesPayload models.AccountStatusesResponse
+	defer messageIndex.Increment()


all places with messageIndex increment were refactored

illia-malachyn · 2025-02-20T12:46:27Z

engine/access/rest/websockets/data_providers/base_provider.go

@@ -15,6 +17,9 @@ type baseDataProvider struct {
 	cancel         context.CancelFunc
 	send           chan<- interface{}
 	subscription   subscription.Subscription
+	// Ensures the closedChan has been closed once.
+	closedFlag sync.Once


Added closedChan channel

illia-malachyn · 2025-02-20T12:47:01Z

engine/access/rest/websockets/data_providers/base_provider.go

+) error {
+	for {
+		select {
+		case <-closedChan:


run() is aware of closedChan and returns nil if it was closed

illia-malachyn · 2025-02-20T12:49:03Z

@peterargue I also pointed out the most important lines of code

…rror-propagation

Guitarheroua · 2025-02-21T16:39:13Z

engine/access/rest/websockets/data_providers/base_provider.go

@@ -33,6 +38,8 @@ func newBaseDataProvider(
 		cancel:         cancel,
 		send:           send,
 		subscription:   subscription,
+		closedFlag:     sync.Once{},
+		closedChan:     make(chan struct{}, 1),


Could we use unbuffered channel here, as it only needed for closing, we do not write to it anywhere?

Guitarheroua · 2025-02-21T16:42:27Z

engine/access/rest/websockets/data_providers/base_provider.go

+
+type sendResponseCallback[T any] func(T) error
+
+func run[T any](


It is self-explentatory, but I would add the function comment anyway.

Guitarheroua · 2025-02-21T17:00:03Z

engine/access/rest/websockets/data_providers/block_digests_provider.go

 		p.subscription,
-		subscription.HandleResponse(p.send, func(b *flow.BlockDigest) (interface{}, error) {


Is this subscription.HandleRespons used anywhere? If not, I think we could remove it.

Guitarheroua

The first round of review - DONE! I have a few small comments.

Properly handle subscription errors in data providers

8f6ca1a

Distinguish between `context.Canceled` errors originating from the streamer and those triggered by the DataProvider’s `Close()` method. Use `wasClosedByClient()` to suppress expected cancellations while propagating unexpected ones

illia-malachyn requested a review from a team as a code owner February 17, 2025 14:22

illia-malachyn requested review from UlyanaAndrukhiv, AndriiDiachuk, Guitarheroua and peterargue February 17, 2025 14:24

illia-malachyn requested a review from durkmurder February 17, 2025 14:28

change reciever name to match previous

acb406d

Guitarheroua previously approved these changes Feb 17, 2025

View reviewed changes

illia-malachyn marked this pull request as draft February 17, 2025 15:02

illia-malachyn changed the title ~~Properly handle subscription errors in data providers~~ [DRAFT] [Access] Properly handle subscription errors in data providers Feb 17, 2025

illia-malachyn added 2 commits February 18, 2025 19:58

Add one shot close channel to all data providers

8230193

We use it to distinguish place where cxt.Canceled error comes from. Also, I refactored each data provider's Run() function. Now it's more readable and clear.

do not spawn unnecessary routine in close()

24e95b8

illia-malachyn marked this pull request as ready for review February 18, 2025 18:15

illia-malachyn added 2 commits February 19, 2025 13:12

remove unused func

a5ef7f3

null blockSinceLastMessage only if we sent a block

be75dbc

illia-malachyn changed the title ~~[DRAFT] [Access] Properly handle subscription errors in data providers~~ [Access] Properly handle subscription errors in data providers Feb 19, 2025

peterargue reviewed Feb 20, 2025

View reviewed changes

illia-malachyn commented Feb 20, 2025

View reviewed changes

illia-malachyn added 2 commits February 21, 2025 14:49

remove unnecessary prevIndex variable

dfa115e

Merge branch 'master' into illia-malachyn/7040-fix-context-canceled-e…

feff7e5

…rror-propagation

Guitarheroua reviewed Feb 21, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Access] Properly handle subscription errors in data providers #7046

[Access] Properly handle subscription errors in data providers #7046

illia-malachyn commented Feb 17, 2025 •

edited

Loading

codecov-commenter commented Feb 17, 2025 •

edited

Loading

Guitarheroua left a comment

illia-malachyn commented Feb 17, 2025 •

edited

Loading

peterargue left a comment

illia-malachyn commented Feb 20, 2025

illia-malachyn Feb 20, 2025

Guitarheroua Feb 21, 2025

illia-malachyn Feb 20, 2025

illia-malachyn Feb 20, 2025

illia-malachyn Feb 20, 2025

illia-malachyn commented Feb 20, 2025

Guitarheroua Feb 21, 2025

Guitarheroua Feb 21, 2025

Guitarheroua Feb 21, 2025

Guitarheroua left a comment •

edited

Loading


		type sendResponseCallback[T any] func(T) error

		func run[T any](

		p.subscription,
		subscription.HandleResponse(p.send, func(b *flow.BlockDigest) (interface{}, error) {

[Access] Properly handle subscription errors in data providers #7046

Are you sure you want to change the base?

[Access] Properly handle subscription errors in data providers #7046

Conversation

illia-malachyn commented Feb 17, 2025 • edited Loading

codecov-commenter commented Feb 17, 2025 • edited Loading

Codecov Report

Guitarheroua left a comment

Choose a reason for hiding this comment

illia-malachyn commented Feb 17, 2025 • edited Loading

peterargue left a comment

Choose a reason for hiding this comment

illia-malachyn commented Feb 20, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

illia-malachyn commented Feb 20, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Guitarheroua left a comment • edited Loading

Choose a reason for hiding this comment

illia-malachyn commented Feb 17, 2025 •

edited

Loading

codecov-commenter commented Feb 17, 2025 •

edited

Loading

illia-malachyn commented Feb 17, 2025 •

edited

Loading

Guitarheroua left a comment •

edited

Loading