DRV-348: Streaming api 2 #127

vadimLF · 2021-01-28T13:18:50Z

Streaming api

import(
    f "github.com/fauna/faunadb-go/v3/faunadb"
    "log"
)
dbClient = f.NewFaunaClient("secret")
value, err := dbClient.Query(
	f.Create(f.Collection("collectionName"), f.Obj{"data": f.Obj{"name": "Magic"}}))
if err != nil {
	panic(err)
}
var ref f.RefV
value.At(f.ObjKey("ref")).Get(&ref)
var sub f.StreamSubscription

subscription = dbClient.Stream(ref)
subscription.Start()
for event := range subscription.Messages() {
	switch event.Type() {
	case f.StartEventT:
		fmt.Println("Start stream")
	case f.VersionEventT:
		fmt.Println("Version event")
	case f.ErrorEventT:
		fmt.Println("Error event")
		subscription.Close()
	}
}

Each started subscription has an event loop running as a goroutine with received stream events sended to a channel
Uses context.Context to impose deadlines on requests instead of relying on HttpClient.Timeout for timeouts.
This change introduces faunaResponse type which stores the http response along with the context and cancel function context.CancelFunc. This allows for different request timeouts for query, set at 60 seconds.
http.NewRequestWithContext is introduced in go 1.13 forcing us to drop support for versions < 1.13 https://golang.org/pkg/net/http/#NewRequestWithContext.
An ErrorEvent is triggered on io.EOF and related server/client errors. The connection is eventually closed and the subscription status updated

`

- added closing channel while closing subscription - added couple of tests

.circleci/config.yml

faunadb/client.go

erickpintor · 2021-02-02T20:56:21Z

faunadb/client.go

+		}
+	}
+
+	if parsedResponse, err = parseJSON(response.Body); err == nil {


Q: Does parseJSON support new-line separated JSONs?

https://github.com/fauna/faunadb-js/blob/master/src/_json.js#L19-L56

json/encoder from go standard library handles this.

Great. Could you please add a test case to prevent regressions?

@erickpintor I've added TestListenToLargeEvents() test for this purpose. Could you pls check ?

So basically, you are right on this issue. I created separate ticket for this - DRV-513.

I'm okay with handling this particular problem at a separate ticket.

faunadb/stream_events.go

erickpintor · 2021-02-02T21:14:03Z

faunadb/stream_subscription.go

+
+// Close eventually closes the stream
+func (sub *StreamSubscription) Close() {
+	if !isClosed(sub.messages) {


What's the relationship between the messages and the closed status of a subscription? Can you rely on the status variable instead?

We need to close event channel in case we are going to close subscription.

My point is: I don't think this isClosed function respects Golang channel semantics: https://play.golang.org/p/gXiR4_5fW8D

As I was referring to in the client receive loop, I think we need a close signal (channel) for this propose.

@erickpintor
my idea with sending close signal was the following:

1:for { 2: select { 3: case <- subscription.closeSignal: 4: subscription.status.Set(StreamConnClosed) 5: close(subscription.messages) 6: default: 7: if val, err := client.parseResponse()... 8:}

Here is the issue
when we obtain data we:
- parsing (line 7)
- event generated (this we are going to send closing signal)
- we are going to parse new portion of data (line 7)
- process logic inside Error event: in this case we are sending close signal
but this signal is never consumed by line 3 since we are at line 7 waiting for new chunk of data

Any idea how to handle this ? or we can just rewrite isClosed since it works.

Pls ignore my previous comment and check new implementation.

erickpintor · 2021-02-02T21:15:31Z

faunadb/stream_test.go

+	for evt := range sub.Messages() {
+		switch evt.Type() {
+		case f.StartEventT:
+			s.NotZero(evt.Txn())
+			s.client.Query(f.Update(ref, f.Obj{"data": f.Obj{"x": time.Now().String()}}))
+		case f.VersionEventT:
+			s.Equal(evt.Type(), f.VersionEventT)
+			sub.Close()
+			wg.Done()
+		case f.ErrorEventT:
+			s.defaultStreamError(evt)
+		}
+	}


This looks really good! 🤘

erickpintor · 2021-02-03T14:15:51Z

faunadb/client.go

+				defer httpResponse.Body.Close()
+				defer response.cncl()
+
+				for {


Q: What instruction in this for loop blocks until there's another message to be read?

There is client.callObserver function in parseResponse.

That's a driver specific callback isn't it? I don't think that's blocking this loop until next event arrives. There are two concerns here: (a) the spinning loop consuming CPU; b) your genera inability to stop this loop at command.

I don't think (a) is happening but I want us to have clarity on which part of it is blocking and waiting for next event to arrive. On (b) I think we need a close signal that we can monitor so that this coroutine can be die when the subscription is closed.

- removed out-of-scope code - removed shadowing issue - changes type for Events from ObjectV to Value - minor fixes

faunadb/client.go

erickpintor · 2021-02-10T12:11:07Z

faunadb/stream_subscription.go

+
+// Close eventually closes the stream
+func (sub *StreamSubscription) Close() {
+	if !isClosed(sub.messages) {


My point is: I don't think this isClosed function respects Golang channel semantics: https://play.golang.org/p/gXiR4_5fW8D

As I was referring to in the client receive loop, I think we need a close signal (channel) for this propose.

- via closing signal

vadimLF · 2021-02-22T18:13:31Z

@erickpintor I added provided signal based solution in this PR. Also I created separate ticket for json streaming issue.

parkhomenko · 2021-02-23T10:18:23Z

hi @erickpintor ,
this PR is here for quite some time, I see that there is an issue that sometimes json that comes from the server is corrupted in terms we need to accumulate it somehow on a client-side. Can we address it as a separate task: https://faunadb.atlassian.net/browse/DRV-513 ?

faunadb/client.go

erickpintor · 2021-02-25T13:42:57Z

faunadb/client.go

+		}
+	}
+
+	if parsedResponse, err = parseJSON(response.Body); err == nil {


I'm okay with handling this particular problem at a separate ticket.

erickpintor · 2021-02-25T13:44:33Z

faunadb/stream_subscription.go

+		go func() {
+			sub.getNext <- true
+		}()


Why do you need a co-routine here?

Since we are using channels for "tracking" closing event I added select statement to startStream() which is waiting for two events closing or obtaining next chunk of data - code

I believe you need a coroutine because these are synchronous channels. If you let it buffer one item (make(chan..., 1)), you can remove the coroutine.

erickpintor · 2021-02-25T13:45:23Z

faunadb/stream_subscription.go

+	}()
+}
+
+func (sub *StreamSubscription) EventsMessages() chan StreamEvent {


Suggested change

func (sub *StreamSubscription) EventsMessages() chan StreamEvent {

func (sub *StreamSubscription) EventsMessages() <-chan StreamEvent {

erickpintor · 2021-02-25T13:46:26Z

faunadb/stream_subscription.go

+func (sub *StreamSubscription) Request() {
+	go func() {
+		sub.getNext <- true
+	}()
+}


Why do you need to actively request the next message? I believe you can rely on channels for reactiveness.

"I believe you can rely on channels for reactiveness." - not sure I understand the way to implement this.
Since we are using channels for "tracking" closing event we also need a way to perform request if there is no closing event in the channel. Also the same behavior is in jvm driver.

This sounds to me that there is something wrong with our model here. Let me restate my understanding:

We have a coroutine in a closed loop reading messages from the network and pushing then to a go channel. Then we have an external command to close the stream which should also kill the coroutine. We've introduced a separate channel to track signal that the subscription is closing which forced us to select between channels in the closed loop. Since reading from the network is a blocking operation, we were then forced to introduce a separate channel managed to the user to signal its code is ready to process the next event.

If that's correct, I believe the usage of select is the issue here. We can move closing resources to a separate coroutine that gets fired on sub.Close(). We'll have to refactor some of the state management for that matter. Please take a look at the following patch. See if it makes sense.

patch.txt

Patch is applied and works as expected. @erickpintor could you please review one more time.

erickpintor · 2021-02-25T13:47:56Z

faunadb/stream_test.go

+
+	subscription = s.client.Stream(ref)
+	subscription.Start()
+	for evt := range subscription.EventsMessages() {


Suggested change

for evt := range subscription.EventsMessages() {

for evt := range subscription.Events() {

I think it looks better if it's just called Events?

We've already have Events in couple of other places. W/hat do you think about StreamEvents ?

erickpintor · 2021-02-26T13:22:28Z

faunadb/stream_subscription.go

+		go func() {
+			sub.getNext <- true
+		}()


I believe you need a coroutine because these are synchronous channels. If you let it buffer one item (make(chan..., 1)), you can remove the coroutine.

erickpintor · 2021-02-26T13:23:29Z

faunadb/stream_test.go

+	for evt := range subscription.StreamEvents() {
+		switch evt.Type() {
+		case f.StartEventT:
+			//s.NotZero(evt.Txn())


Suggested change

//s.NotZero(evt.Txn())

faunadb/stream_test.go

erickpintor · 2021-02-26T13:38:55Z

faunadb/client.go

+						break
+					}
+					subscription.eventsMessages <- ErrorEvent{
+						txn: startTime.Unix(),


Fauna errors will report a transaction timestamp, we should use that.

erickpintor · 2021-02-26T14:47:21Z

faunadb/stream_subscription.go

+func (sub *StreamSubscription) Request() {
+	go func() {
+		sub.getNext <- true
+	}()
+}


This sounds to me that there is something wrong with our model here. Let me restate my understanding:

We have a coroutine in a closed loop reading messages from the network and pushing then to a go channel. Then we have an external command to close the stream which should also kill the coroutine. We've introduced a separate channel to track signal that the subscription is closing which forced us to select between channels in the closed loop. Since reading from the network is a blocking operation, we were then forced to introduce a separate channel managed to the user to signal its code is ready to process the next event.

If that's correct, I believe the usage of select is the issue here. We can move closing resources to a separate coroutine that gets fired on sub.Close(). We'll have to refactor some of the state management for that matter. Please take a look at the following patch. See if it makes sense.

patch.txt

erickpintor

One last change and I think this is good to go.

faunadb/client.go

erickpintor · 2021-03-02T15:25:23Z

faunadb/client.go

+					break
+				}
+				subscription.events <- ErrorEvent{
+					txn: startTime.Unix(),


We can't introduce synthetic transaction times. That can lead to consistency issues if users rely on this marker for their domain logic. In this cases, it's best to not report back the a transaction time (you this line and let go assign the vars default value 0).

This change has to be replicated to all ErrorEvent initializations. As a guideline, pick the transaction time from the server. If not possible, remove the txn and let it be zero.

Let's remove timestamp from manually generated events.
Done.

…eaming-api-2

erickpintor

🐑 it! Great job!

vadimLF changed the title ~~Streaming api 2~~ DRV-348: Streaming api 2 Jan 28, 2021

vadimLF requested review from erickpintor, marrony and parkhomenko January 28, 2021 14:33

Trevor Sibanda and others added 6 commits February 1, 2021 11:33

Introduce streaming api

599e8bf

Initial commit

21ed569

Added Field as parameter

ba1f8d6

- removed sending message to channel in case of badRequest

d8a5ca0

- added closing channel while closing subscription - added couple of tests

Added more tests

80028f5

Added isClosed func for channels

86fe154

vadimLF force-pushed the streaming-api-2 branch from 3820043 to 86fe154 Compare February 1, 2021 09:36

Handle history_rewrite correctly

365b380

erickpintor reviewed Feb 2, 2021

View reviewed changes

erickpintor reviewed Feb 3, 2021

View reviewed changes

vadimLF added 7 commits February 5, 2021 14:02

Fixes according to review comments

b6a8b50

- removed out-of-scope code - removed shadowing issue - changes type for Events from ObjectV to Value - minor fixes

Removed changes in config.yml

6d225a4

Fixed issue with failed build

60d3028

Fixed failed test

b2688ff

Fixed failed tests

90868ec

Drop old go version support for circleCI

bc62226

Minor fix

d5f03c5

marrony mentioned this pull request Feb 9, 2021

DRV-348: Introduce streaming api #121

Closed

erickpintor reviewed Feb 10, 2021

View reviewed changes

vadimLF added 3 commits February 10, 2021 15:50

Minor fixes

9dbe04d

Added stream test for large files

51d6789

Implemented new version of closing subscriptions:

647680e

- via closing signal

vadimLF marked this pull request as ready for review February 23, 2021 10:19

erickpintor requested changes Feb 25, 2021

View reviewed changes

Fix review comments

8d8e6a2

erickpintor reviewed Feb 26, 2021

View reviewed changes

vadimLF added 2 commits March 1, 2021 18:53

Apply patch from Erick, code cleanup

2649b46

Minor fix

60258fc

erickpintor requested changes Mar 2, 2021

View reviewed changes

vadimLF added 2 commits March 2, 2021 18:01

Removed timestamp from manually generated Events

7f839b9

Merge branch 'master' of https://github.com/fauna/faunadb-go into str…

c92ca8a

…eaming-api-2

erickpintor approved these changes Mar 2, 2021

View reviewed changes

parkhomenko approved these changes Mar 3, 2021

View reviewed changes

vadimLF merged commit feb1acc into fauna:master Mar 3, 2021

vadimLF deleted the streaming-api-2 branch March 3, 2021 08:07

	func (sub *StreamSubscription) EventsMessages() chan StreamEvent {
	func (sub *StreamSubscription) EventsMessages() <-chan StreamEvent {

	for evt := range subscription.EventsMessages() {
	for evt := range subscription.Events() {

DRV-348: Streaming api 2 #127

DRV-348: Streaming api 2 #127

Conversation

vadimLF commented Jan 28, 2021 • edited Loading

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

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

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

erickpintor Feb 10, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vadimLF commented Feb 22, 2021

parkhomenko commented Feb 23, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vadimLF Feb 25, 2021 • edited Loading

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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

erickpintor left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

erickpintor left a comment

Choose a reason for hiding this comment

vadimLF commented Jan 28, 2021 •

edited

Loading

erickpintor Feb 10, 2021 •

edited

Loading

vadimLF Feb 25, 2021 •

edited

Loading