Merge pull request #356 from gotd/fix/peer-storage-marshal

fix(storage): Peer marshaling

.github/workflows/ci.yml

@@ -16,24 +16,24 @@ jobs:
       matrix:
         flags: [""]
         go:
-          - 1.19
+          - 1.20.x
         arch:
           - amd64
         runner:
           - ubuntu-latest
           - macos-latest
         include:
           - arch: 386
-            go: 1.19
+            go: 1.20.x
             runner: ubuntu-latest
 
           - arch: amd64
             runner: windows-latest
-            go: 1.19
+            go: 1.20.x
             flags: "-p=1"
 
           - arch: amd64
-            go: 1.19
+            go: 1.20.x
             runner: ubuntu-latest
             flags: "-race"
     steps:

.github/workflows/coverage.yml

@@ -14,7 +14,7 @@ jobs:
       - name: Install Go
         uses: actions/[email protected]
         with:
-          go-version: 1.19
+          go-version: 1.20.x
 
       - name: Get Go environment
         id: go-env

.github/workflows/e2e.yml

@@ -144,7 +144,7 @@ jobs:
       - name: Install Go
         uses: actions/[email protected]
         with:
-          go-version: 1.19
+          go-version: 1.20.x
 
       - name: Get Go environment
         id: go-env

.github/workflows/lint.yml

@@ -18,7 +18,7 @@ jobs:
       - name: Install Go
         uses: actions/[email protected]
         with:
-          go-version: 1.19
+          go-version: 1.20.x
 
       - name: Lint
         uses: golangci/[email protected]
@@ -36,7 +36,7 @@ jobs:
       - name: Install Go
         uses: actions/[email protected]
         with:
-          go-version: 1.19
+          go-version: 1.20.x
 
       - name: Get Go environment
         id: go-env
@@ -69,7 +69,7 @@ jobs:
       - name: Install Go
         uses: actions/[email protected]
         with:
-          go-version: 1.19
+          go-version: 1.20.x
 
       - name: Get Go environment
         id: go-env

.golangci.yml

@@ -39,7 +39,6 @@ linters-settings:
 linters:
   disable-all: true
   enable:
-    - deadcode
     - depguard
     - dogsled
     - dupl
@@ -57,17 +56,14 @@ linters:
     - lll
     - misspell
     - nakedret
-    - staticcheck
-    - structcheck
     - stylecheck
     - typecheck
     - unconvert
     - unparam
     - unused
-    - varcheck
     - whitespace
     - gochecknoglobals
-    - gocognit
+#    - gocognit
 
   # Do not enable:
   # - wsl       (too opinionated about newlines)

bbolt/peer_storage.go

@@ -36,6 +36,7 @@ func (p *bboltIterator) Close() error {
 }
 
 func (p *bboltIterator) Next(ctx context.Context) bool {
+Next:
 	k, v := p.iter.Next()
 	if v == nil {
 		return false
@@ -53,7 +54,10 @@ func (p *bboltIterator) Next(ctx context.Context) bool {
 	}
 
 	if err := json.Unmarshal(v, &p.value); err != nil {
-		p.lastErr = errors.Errorf("unmarshal: %w", err)
+		if errors.Is(err, storage.ErrPeerUnmarshalMustInvalidate) {
+			goto Next // skip
+		}
+		p.lastErr = errors.Wrap(err, "unmarshal")
 		return false
 	}
 
@@ -136,6 +140,9 @@ func (s PeerStorage) Find(ctx context.Context, key storage.PeerKey) (p storage.P
 		}
 
 		if err := json.Unmarshal(data, &p); err != nil {
+			if errors.Is(err, storage.ErrPeerUnmarshalMustInvalidate) {
+				return storage.ErrPeerNotFound
+			}
 			return errors.Errorf("unmarshal: %w", err)
 		}
 		return nil
@@ -167,6 +174,9 @@ func (s PeerStorage) Resolve(ctx context.Context, key string) (p storage.Peer, r
 		}
 
 		if err := json.Unmarshal(data, &p); err != nil {
+			if errors.Is(err, storage.ErrPeerUnmarshalMustInvalidate) {
+				return storage.ErrPeerNotFound
+			}
 			return errors.Errorf("unmarshal: %w", err)
 		}
 		return nil

go.mod

@@ -1,13 +1,14 @@
 module github.com/gotd/contrib
 
-go 1.18
+go 1.20
 
 require (
 	github.com/beevik/ntp v0.3.0
 	github.com/cenkalti/backoff/v4 v4.2.0
 	github.com/cockroachdb/pebble v0.0.0-20220107203702-aa376a819bf6
 	github.com/gen2brain/dlgs v0.0.0-20211108104213-bade24837f0b
 	github.com/go-faster/errors v0.6.1
+	github.com/go-faster/jx v1.0.0
 	github.com/go-redis/redis/v8 v8.11.5
 	github.com/gotd/neo v0.1.5
 	github.com/gotd/td v0.79.0
@@ -42,7 +43,6 @@ require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/dustin/go-humanize v1.0.1 // indirect
-	github.com/go-faster/jx v1.0.0 // indirect
 	github.com/go-faster/xor v0.3.0 // indirect
 	github.com/gogo/protobuf v1.3.1 // indirect
 	github.com/golang/protobuf v1.5.3 // indirect

internal/tests/storage.go

@@ -114,6 +114,14 @@ func TestPeerStorage(t *testing.T, st storage.PeerStorage) {
 		}
 
 		a.GreaterOrEqual(len(peers), 6)
-		a.Contains(peers, p)
+		var found bool
+		for _, vp := range peers {
+			if vp.Key != p.Key {
+				continue
+			}
+			found = true
+			break
+		}
+		a.True(found, "should contain")
 	})
 }

middleware/floodwait/waiter.go

@@ -37,6 +37,12 @@ type Waiter struct {
 	tick       time.Duration
 	maxWait    time.Duration
 	maxRetries int
+	callback   func(ctx context.Context, wait FloodWait)
+}
+
+// FloodWait event.
+type FloodWait struct {
+	Duration time.Duration
 }
 
 // NewWaiter returns a new invoker that waits on the flood wait errors.
@@ -47,9 +53,17 @@ func NewWaiter() *Waiter {
 		tick:       defaultTick,
 		maxWait:    defaultMaxWait,
 		maxRetries: defaultMaxRetries,
+		callback:   func(ctx context.Context, wait FloodWait) {},
 	}
 }
 
+// WithCallback sets callback for flood wait event.
+func (w *Waiter) WithCallback(f func(ctx context.Context, wait FloodWait)) *Waiter {
+	w = w.clone()
+	w.callback = f
+	return w
+}
+
 // clone returns a copy of the Waiter.
 func (w *Waiter) clone() *Waiter {
 	return &Waiter{
@@ -136,6 +150,11 @@ func (w *Waiter) send(s scheduled) (bool, error) {
 		return true, err
 	}
 
+	// Notify about flood wait.
+	w.callback(s.request.ctx, FloodWait{
+		Duration: d,
+	})
+
 	s.request.retry++
 
 	if max := w.maxRetries; max != 0 && s.request.retry > max {

pebble/peer_storage.go

@@ -158,6 +158,9 @@ func (s PeerStorage) Find(ctx context.Context, key storage.PeerKey) (_ storage.P
 
 	var b storage.Peer
 	if err := json.Unmarshal(data, &b); err != nil {
+		if errors.Is(err, storage.ErrPeerUnmarshalMustInvalidate) {
+			return storage.Peer{}, storage.ErrPeerNotFound
+		}
 		return storage.Peer{}, errors.Errorf("unmarshal: %w", err)
 	}
 
@@ -203,6 +206,9 @@ func (s PeerStorage) Resolve(ctx context.Context, key string) (_ storage.Peer, r
 
 	var b storage.Peer
 	if err := json.Unmarshal(data, &b); err != nil {
+		if errors.Is(err, storage.ErrPeerUnmarshalMustInvalidate) {
+			return storage.Peer{}, storage.ErrPeerNotFound
+		}
 		return storage.Peer{}, errors.Errorf("unmarshal: %w", err)
 	}