Repository: yorkie-team/yorkie
Pull request #479 (2023-03-02 12:06:51)

Lock watchDocuments depending on the client
<!--  Thanks for sending a pull request! -->

**What this PR does / why we need it**:

If a client requests to attach to multiple documents simultaneously, 
the existing connection will be disconnected and reconnected. 

Depending on the order of the published events, problems may occur on the client. 
If the `watched` event is published before the `unwatched` event, the test will fail 
since the final state is `watched`, but the `unwatched` event arrives later. 
(please refer to the [yorkie-js-sdk test](https://github.com/yorkie-team/yorkie-js-sdk/pull/464#issuecomment-1447547066)). 

<img width="600" alt="image" src="https://user-images.githubusercontent.com/81357083/222397680-8479c102-56c9-4af2-9601-255df4e87c4b.png">

This image shows the server log when the `watched` event is published first.

To resolve this, I added a lock for each client in the `watchDocument`.
This ensures that a lock is set when a client establishes a stream and released when the client unwatches.

**Which issue(s) this PR fixes**:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
-->
Fixes https://github.com/yorkie-team/yorkie-js-sdk/pull/464#issuecomment-1447547066

**Special notes for your reviewer**:

**Does this PR introduce a user-facing change?**:
<!--
If no, just write "NONE" in the release-note block below.
If yes, a release note is required:
Enter your extended release note in the block below. If the PR requires additional action from users switching to the new release, include the string "action required".
-->
```release-note

```

**Additional documentation**:

<!--
This section can be blank if this pull request does not require a release note.

Please use the following format for linking documentation:
- [Usage]: <link>
- [Other doc]: <link>
-->
```docs

```

**Checklist**:
- [x] Added relevant tests or not required
- [x] Didn't break anything


codecov[bot]: # [Codecov](https://codecov.io/gh/yorkie-team/yorkie/pull/479?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yorkie-team) Report
> Merging [#479](https://codecov.io/gh/yorkie-team/yorkie/pull/479?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yorkie-team) (79e2f9c) into [main](https://codecov.io/gh/yorkie-team/yorkie/commit/bc3f59b6b1c7370baebb7c803b44c7565002a50c?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yorkie-team) (bc3f59b) will **decrease** coverage by `0.09%`.
> The diff coverage is `0.00%`.

```diff
@@            Coverage Diff             @@
##             main     #479      +/-   ##
==========================================
- Coverage   46.56%   46.48%   -0.09%     
==========================================
  Files          69       69              
  Lines        5788     5798      +10     
==========================================
  Hits         2695     2695              
- Misses       2785     2795      +10     
  Partials      308      308              
```


| [Impacted Files](https://codecov.io/gh/yorkie-team/yorkie/pull/479?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yorkie-team) | Coverage Δ | |
|---|---|---|
| [server/backend/sync/memory/pubsub.go](https://codecov.io/gh/yorkie-team/yorkie/pull/479?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yorkie-team#diff-c2VydmVyL2JhY2tlbmQvc3luYy9tZW1vcnkvcHVic3ViLmdv) | `47.11% <0.00%> (-0.46%)` | :arrow_down: |
| [server/rpc/yorkie\_server.go](https://codecov.io/gh/yorkie-team/yorkie/pull/479?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yorkie-team#diff-c2VydmVyL3JwYy95b3JraWVfc2VydmVyLmdv) | `46.73% <0.00%> (-1.42%)` | :arrow_down: |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yorkie-team). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yorkie-team)


Patch:
diff --git a/pkg/document/key/key.go b/pkg/document/key/key.go
index 8001a173..4595e8d3 100644
--- a/pkg/document/key/key.go
+++ b/pkg/document/key/key.go
@@ -19,6 +19,7 @@ package key
 
 import (
 	errors "errors"
+	"strings"
 
 	"github.com/yorkie-team/yorkie/internal/validation"
 )
@@ -49,3 +50,15 @@ func (k Key) Validate() error {
 
 	return nil
 }
+
+// Join concatenates all the keys in slice into a single string.
+func Join(keys []Key) string {
+	var s strings.Builder
+	for i, key := range keys {
+		s.WriteString(key.String())
+		if i != len(keys)-1 {
+			s.WriteString(",")
+		}
+	}
+	return s.String()
+}
diff --git a/server/backend/sync/memory/pubsub.go b/server/backend/sync/memory/pubsub.go
index b28d554a..3a6fe081 100644
--- a/server/backend/sync/memory/pubsub.go
+++ b/server/backend/sync/memory/pubsub.go
@@ -93,7 +93,7 @@ func (m *PubSub) Subscribe(
 	if logging.Enabled(zap.DebugLevel) {
 		logging.From(ctx).Debugf(
 			`Subscribe(%s,%s) Start`,
-			keys[0],
+			key.Join(keys),
 			subscriber.ID.String(),
 		)
 	}
@@ -115,7 +115,7 @@ func (m *PubSub) Subscribe(
 	if logging.Enabled(zap.DebugLevel) {
 		logging.From(ctx).Debugf(
 			`Subscribe(%s,%s) End`,
-			keys[0],
+			key.Join(keys),
 			subscriber.ID.String(),
 		)
 	}
@@ -148,7 +148,7 @@ func (m *PubSub) Unsubscribe(
 	if logging.Enabled(zap.DebugLevel) {
 		logging.From(ctx).Debugf(
 			`Unsubscribe(%s,%s) Start`,
-			docKeys[0].String(),
+			key.Join(docKeys),
 			sub.SubscriberID(),
 		)
 	}
@@ -170,7 +170,7 @@ func (m *PubSub) Unsubscribe(
 	if logging.Enabled(zap.DebugLevel) {
 		logging.From(ctx).Debugf(
 			`Unsubscribe(%s,%s) End`,
-			docKeys[0].String(),
+			key.Join(docKeys),
 			sub.SubscriberID(),
 		)
 	}
@@ -200,7 +200,8 @@ func (m *PubSub) Publish(
 
 				if logging.Enabled(zap.DebugLevel) {
 					logging.From(ctx).Debugf(
-						`Publish(%s,%s) to %s`,
+						`Publish %s(%s,%s) to %s`,
+						event.Type,
 						k,
 						publisherID.String(),
 						sub.SubscriberID(),
diff --git a/server/rpc/yorkie_server.go b/server/rpc/yorkie_server.go
index 0d465bfc..79141d5d 100644
--- a/server/rpc/yorkie_server.go
+++ b/server/rpc/yorkie_server.go
@@ -18,6 +18,7 @@ package rpc
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/yorkie-team/yorkie/api/converter"
 	"github.com/yorkie-team/yorkie/api/types"
@@ -392,6 +393,22 @@ func (s *yorkieServer) WatchDocuments(
 		return err
 	}
 
+	locker, err := s.backend.Coordinator.NewLocker(
+		stream.Context(),
+		sync.NewKey(fmt.Sprintf("watchdocs-%s", cli.ID.String())),
+	)
+	if err != nil {
+		return err
+	}
+	if err := locker.Lock(stream.Context()); err != nil {
+		return err
+	}
+	defer func() {
+		if err := locker.Unlock(context.Background()); err != nil {
+			logging.DefaultLogger().Error(err)
+		}
+	}()
+
 	subscription, peersMap, err := s.watchDocs(stream.Context(), *cli, docKeys)
 	if err != nil {
 		logging.From(stream.Context()).Error(err)


Test patch:
diff --git a/pkg/document/key/key_test.go b/pkg/document/key/key_test.go
index 91858e1b..6914586e 100644
--- a/pkg/document/key/key_test.go
+++ b/pkg/document/key/key_test.go
@@ -54,4 +54,10 @@ func TestKey(t *testing.T) {
 		err = Key("inv").Validate()
 		assert.Equal(t, err, ErrInvalidKey, "less than 4 characters is not allowed")
 	})
+
+	t.Run("join key test", func(t *testing.T) {
+		keys := []Key{Key("key1"), Key("key2"), Key("key3")}
+		result := Join(keys)
+		assert.Equal(t, "key1,key2,key3", result)
+	})
 }