Skip to content

Commit 83d72ef

Browse files
anbskyanbsky
committed
Improve cache retrieval and logging
1 parent 0fd57a9 commit 83d72ef

File tree

5 files changed

+98
-46
lines changed

5 files changed

+98
-46
lines changed

‎app/query/cache.go‎

Lines changed: 23 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -56,40 +56,49 @@ func (c *QueryCache) Retrieve(query *Query, getter func() (any, error)) (*Cached
5656
if err != nil {
5757
if !errors.Is(err, &store.NotFound{}) {
5858
metrics.SturdyQueryCacheErrorCount.WithLabelValues(cacheReq.Method).Inc()
59-
return nil, fmt.Errorf("failed to cache.get: %w", err)
59+
return nil, fmt.Errorf("error during cache.get: %w", err)
6060
}
6161

6262
metrics.SturdyQueryCacheMissCount.WithLabelValues(cacheReq.Method).Inc()
6363

6464
if getter == nil {
65-
log.Warnf("nil getter provided for %s", query.Method())
65+
log.Warnf("nil getter provided for %s", cacheReq.Method)
6666
metrics.SturdyQueryCacheErrorCount.WithLabelValues(cacheReq.Method).Inc()
67-
return nil, errors.New("cache miss with no object getter provided")
67+
return nil, nil
6868
}
6969

70-
log.Infof("cold cache retrieval for %s", query.Method())
70+
// Cold object retrieval after cache miss
71+
log.Infof("cold object retrieval for %s [%s]", cacheReq.Method, cacheReq.GetCacheKey())
7172
obj, err, _ := c.singleflight.Do(cacheReq.GetCacheKey(), getter)
7273
if err != nil {
7374
metrics.SturdyQueryCacheErrorCount.WithLabelValues(cacheReq.Method).Inc()
74-
return nil, fmt.Errorf("failed to call object getter: %w", err)
75+
return nil, fmt.Errorf("error calling getter: %w", err)
7576
}
77+
7678
res, ok := obj.(*jsonrpc.RPCResponse)
7779
if !ok {
7880
return nil, errors.New("unknown type returned by getter")
7981
}
82+
8083
cacheResp := &CachedResponse{Result: res.Result, Error: res.Error}
81-
err = c.cache.Set(
82-
ctx, cacheReq, cacheResp,
83-
store.WithExpiration(cacheReq.Expiration()),
84-
store.WithTags(cacheReq.Tags()),
85-
)
86-
if err != nil {
87-
metrics.SturdyQueryCacheErrorCount.WithLabelValues(cacheReq.Method).Inc()
88-
monitor.ErrorToSentry(fmt.Errorf("failed to cache.set: %w", err), map[string]string{"method": cacheReq.Method})
89-
return nil, fmt.Errorf("failed to cache.set: %w", err)
84+
if res.Error != nil {
85+
log.Debugf("rpc error received (%s), not caching", cacheReq.Method)
86+
} else {
87+
err = c.cache.Set(
88+
ctx, cacheReq, cacheResp,
89+
store.WithExpiration(cacheReq.Expiration()),
90+
store.WithTags(cacheReq.Tags()),
91+
)
92+
if err != nil {
93+
metrics.SturdyQueryCacheErrorCount.WithLabelValues(cacheReq.Method).Inc()
94+
monitor.ErrorToSentry(fmt.Errorf("error during cache.set: %w", err), map[string]string{"method": cacheReq.Method})
95+
return cacheResp, fmt.Errorf("error during cache.set: %w", err)
96+
}
9097
}
98+
9199
return cacheResp, nil
92100
}
101+
log.Debugf("cache hit for %s [%s]", cacheReq.Method, cacheReq.GetCacheKey())
93102
cacheResp, ok := hit.(*CachedResponse)
94103
if !ok {
95104
metrics.SturdyQueryCacheErrorCount.WithLabelValues(cacheReq.Method).Inc()
@@ -157,12 +166,10 @@ func preflightCacheHook(caller *Caller, ctx context.Context) (*jsonrpc.RPCRespon
157166
}
158167
query := QueryFromContext(ctx)
159168
cachedResp, err := caller.Cache.Retrieve(query, func() (any, error) {
160-
log.Debugf("cache miss, calling %s", query.Method())
161169
return caller.SendQuery(ctx, query)
162170
})
163171
if err != nil {
164172
return nil, rpcerrors.NewSDKError(err)
165173
}
166-
log.Debugf("cache hit for %s", query.Method())
167174
return cachedResp.RPCResponse(query.Request.ID), nil
168175
}

‎app/query/cache_test.go‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
package query
2+
3+
import (
4+
"fmt"
5+
"testing"
6+
7+
"github.com/stretchr/testify/assert"
8+
"github.com/stretchr/testify/require"
9+
"github.com/ybbus/jsonrpc"
10+
)
11+
12+
func TestGetCacheKey(t *testing.T) {
13+
assert := assert.New(t)
14+
require := require.New(t)
15+
assert.Equal(1, 1)
16+
seen := map[string]bool{}
17+
params := []map[string]any{{}, {"uri": "what"}, {"uri": "odysee"}, nil}
18+
genCacheKey := func(params map[string]any) string {
19+
req := jsonrpc.NewRequest(MethodResolve, params)
20+
query, err := NewQuery(req, "")
21+
require.NoError(err)
22+
cacheReq := CacheRequest{
23+
Method: query.Method(),
24+
Params: query.Params(),
25+
}
26+
return cacheReq.GetCacheKey()
27+
}
28+
for _, p := range params {
29+
t.Run(fmt.Sprintf("%+v", p), func(t *testing.T) {
30+
cacheKey := genCacheKey(p)
31+
assert.Len(cacheKey, 32)
32+
assert.NotContains(seen, cacheKey)
33+
seen[cacheKey] = true
34+
})
35+
}
36+
assert.Contains(seen, genCacheKey(params[1]))
37+
}

‎app/query/caller.go‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -157,8 +157,8 @@ func (c *Caller) Endpoint() string {
157157
return c.endpoint
158158
}
159159

160-
// Call method forwards a JSON-RPC request to the lbrynet server.
161-
// It returns a response that is ready to be sent back to the JSON-RPC client as is.
160+
// Call method takes JSON-RPC request through a set of hooks and forwards it to lbrynet server.
161+
// It returns a response that is ready to be sent back to the JSON-RPC client.
162162
func (c *Caller) Call(ctx context.Context, req *jsonrpc.RPCRequest) (*jsonrpc.RPCResponse, error) {
163163
origin := OriginFromContext(ctx)
164164
metrics.ProxyCallCounter.WithLabelValues(req.Method, c.Endpoint(), origin).Inc()
@@ -171,8 +171,6 @@ func (c *Caller) Call(ctx context.Context, req *jsonrpc.RPCRequest) (*jsonrpc.RP
171171
return res, err
172172
}
173173

174-
// Call method forwards a JSON-RPC request to the lbrynet server.
175-
// It returns a response that is ready to be sent back to the JSON-RPC client as is.
176174
func (c *Caller) call(ctx context.Context, req *jsonrpc.RPCRequest) (*jsonrpc.RPCResponse, error) {
177175
if c.endpoint == "" {
178176
return nil, errors.Err("cannot call blank endpoint")
@@ -207,6 +205,7 @@ func (c *Caller) call(ctx context.Context, req *jsonrpc.RPCRequest) (*jsonrpc.RP
207205
return c.SendQuery(ctx, q)
208206
}
209207

208+
// SendQuery is where the actual RPC call happens, bypassing all hooks and retrying in case of "wallet not loaded" errors.
210209
func (c *Caller) SendQuery(ctx context.Context, q *Query) (*jsonrpc.RPCResponse, error) {
211210
var (
212211
r *jsonrpc.RPCResponse

‎app/query/caller_test.go‎

Lines changed: 19 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -381,27 +381,11 @@ func TestCaller_CloneWithoutHook(t *testing.T) {
381381

382382
func TestCaller_CallCachingResponses(t *testing.T) {
383383
var err error
384+
384385
srv := test.MockHTTPServer(nil)
385386
defer srv.Close()
386-
srv.NextResponse <- `
387-
{
388-
"jsonrpc": "2.0",
389-
"result": {
390-
"blocked": {
391-
"channels": [],
392-
"total": 0
393-
},
394-
"items": [
395-
{
396-
"address": "bHz3LpVcuadmbK8g6VVUszF9jjH4pxG2Ct",
397-
"amount": "0.5",
398-
"canonical_url": "lbry://@lbry#3f/youtube-is-over-lbry-odysee-are-here#4"
399-
}
400-
]
401-
},
402-
"id": 0
403-
}
404-
`
387+
388+
srv.NextResponse <- resolveResponseFree
405389

406390
c := NewCaller(srv.URL, 0)
407391

@@ -410,21 +394,30 @@ func TestCaller_CallCachingResponses(t *testing.T) {
410394
c.Cache = NewQueryCache(cache)
411395
require.NoError(t, err)
412396

413-
rpcReq := jsonrpc.NewRequest("claim_search", map[string]any{"urls": "what"})
397+
rpcReq := jsonrpc.NewRequest("resolve", map[string]any{"urls": "what"})
414398
rpcResponse, err := c.Call(bgctx(), rpcReq)
415399
require.NoError(t, err)
416400
assert.Nil(t, rpcResponse.Error)
417401

418-
time.Sleep(1 * time.Second)
419-
q, _ := NewQuery(rpcReq, "")
402+
expResponse, err := decodeResponse(resolveResponseFree)
403+
require.NoError(t, err)
404+
assert.EqualValues(t, expResponse.Result, rpcResponse.Result)
420405

421-
cResp, err := c.Cache.Retrieve(q, nil)
406+
srv.NextResponse <- resolveResponseCouldntFind
407+
408+
rpcReq2 := jsonrpc.NewRequest("resolve", map[string]any{"urls": "one"})
409+
rpcResponse2, err := c.Call(bgctx(), rpcReq2)
422410
require.NoError(t, err)
423-
assert.NotNil(t, cResp.Result)
411+
assert.Nil(t, rpcResponse.Error)
412+
413+
expResponse2, err := decodeResponse(resolveResponseCouldntFind)
414+
require.NoError(t, err)
415+
assert.Nil(t, rpcResponse2.Error)
416+
assert.EqualValues(t, expResponse2.Result, rpcResponse2.Result)
417+
424418
}
425419

426420
func TestCaller_CallNotCachingErrors(t *testing.T) {
427-
t.SkipNow()
428421
var err error
429422
srv := test.MockHTTPServer(nil)
430423
defer srv.Close()
@@ -453,7 +446,7 @@ func TestCaller_CallNotCachingErrors(t *testing.T) {
453446
q, err := NewQuery(rpcReq, "")
454447
require.NoError(t, err)
455448

456-
cResp, err := c.Cache.Retrieve(q, func() (any, error) { return nil, nil })
449+
cResp, err := c.Cache.Retrieve(q, nil)
457450
require.NoError(t, err)
458451
assert.Nil(t, cResp)
459452
}

‎app/query/testing.go‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
package query
2+
3+
import (
4+
"encoding/json"
5+
"strings"
6+
7+
"github.com/ybbus/jsonrpc"
8+
)
9+
10+
func decodeResponse(r string) (*jsonrpc.RPCResponse, error) {
11+
decoder := json.NewDecoder(strings.NewReader(r))
12+
decoder.DisallowUnknownFields()
13+
decoder.UseNumber()
14+
response := &jsonrpc.RPCResponse{}
15+
return response, decoder.Decode(response)
16+
}

0 commit comments

Comments
 (0)