-
Notifications
You must be signed in to change notification settings - Fork 2.6k
Expand file tree
/
Copy pathcsc_commands.go
More file actions
257 lines (237 loc) · 8.45 KB
/
Copy pathcsc_commands.go
File metadata and controls
257 lines (237 loc) · 8.45 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
package redis
import (
"bytes"
"strconv"
"strings"
"github.com/redis/go-redis/v9/internal/proto"
)
// defaultCacheableCommands is the allow-list of read-only, deterministic
// commands whose responses may be stored in the client-side cache. Keys are
// lowercase to match baseCmd.Name() on the hot path.
var defaultCacheableCommands = map[string]struct{}{
// String commands
"get": {}, "mget": {}, "getbit": {}, "getrange": {},
"strlen": {}, "substr": {},
// Hash commands
"hget": {}, "hgetall": {}, "hmget": {},
"hkeys": {}, "hvals": {}, "hlen": {},
"hexists": {}, "hstrlen": {},
// List commands
"lindex": {}, "llen": {}, "lpos": {}, "lrange": {},
// Set commands
"scard": {}, "sismember": {}, "smembers": {}, "smismember": {},
"sdiff": {}, "sinter": {}, "sintercard": {}, "sunion": {},
// Sorted-set commands
"zcard": {}, "zcount": {}, "zlexcount": {}, "zmscore": {},
"zrange": {}, "zrangebylex": {}, "zrangebyscore": {},
"zrank": {}, "zrevrange": {}, "zrevrangebylex": {},
"zrevrangebyscore": {}, "zrevrank": {}, "zscore": {},
"zdiff": {}, "zinter": {}, "zunion": {},
// Bit commands
"bitcount": {}, "bitfield_ro": {}, "bitpos": {},
// Key/generic commands
"exists": {}, "type": {}, "sort_ro": {}, "lcs": {},
// Geo commands
"geodist": {}, "geohash": {}, "geopos": {}, "geosearch": {},
"georadiusbymember_ro": {}, "georadius_ro": {},
// Stream commands. XREAD is deliberately excluded: it supports BLOCK, and
// its $/+ IDs are state-relative, so identical args are not deterministic.
// XPENDING is excluded for the same class of reason: its extended form
// returns wall-clock-relative idle times and its IDLE filter is
// time-dependent, so identical args yield different correct results with
// no key modification (and therefore no invalidation).
"xlen": {}, "xrange": {}, "xrevrange": {},
// JSON (RedisJSON) commands
"json.get": {}, "json.mget": {}, "json.arrindex": {}, "json.arrlen": {},
"json.objkeys": {}, "json.objlen": {}, "json.resp": {},
"json.strlen": {}, "json.type": {},
// TimeSeries commands
"ts.get": {}, "ts.info": {}, "ts.range": {}, "ts.revrange": {},
}
// isCacheable reports whether cmd is eligible for client-side caching: its
// name is on the allow-list and it operates on at least one key.
func isCacheable(cmd Cmder) bool {
// Commands such as RawWriteToCmd stream replies directly to an io.Writer.
// Capturing their replies for CSC would buffer the entire response first,
// defeating their streaming and allocation guarantees.
if cmd.NoRetry() {
return false
}
if _, ok := defaultCacheableCommands[cmd.Name()]; !ok {
return false
}
// SORT_RO ... BY/GET reads pattern keys that extractRedisKeys can't
// enumerate, so its invalidations would be dropped and the result go stale.
// Plain SORT_RO is fine.
if cmd.Name() == "sort_ro" && sortROHasByGet(cmd) {
return false
}
return cmdFirstKeyPosWithInfo(cmd, nil) != 0
}
// sortROHasByGet reports whether a SORT_RO invocation uses BY or GET
// (case-insensitive), scanning past the command name and key. stringArg
// normalizes string, *string, and []byte tokens.
func sortROHasByGet(cmd Cmder) bool {
for i := 2; i < len(cmd.Args()); i++ {
if s := cmd.stringArg(i); strings.EqualFold(s, "by") || strings.EqualFold(s, "get") {
return true
}
}
return false
}
// isClientTrackingCmd reports whether cmd is a CLIENT TRACKING subcommand (any
// mode: ON, OFF, or with options). Name and stringArg normalize string,
// *string, and []byte arguments.
func isClientTrackingCmd(cmd Cmder) bool {
return cmd.Name() == "client" && strings.EqualFold(cmd.stringArg(1), "tracking")
}
// isSelectCmd reports whether cmd changes the selected database on its
// connection. CSC keys are namespaced with Options.DB, so a runtime SELECT
// would make the connection's actual database diverge from the cache namespace.
func isSelectCmd(cmd Cmder) bool {
return cmd.Name() == "select"
}
// isAuthCmd reports whether cmd changes the authenticated user on its
// connection. The cache namespace is fixed from Options.Username, so runtime
// authentication would make the connection identity diverge from it.
func isAuthCmd(cmd Cmder) bool {
return cmd.Name() == "auth"
}
// isProtocolChangingHelloCmd reports whether HELLO includes a protocol version
// (and can therefore switch a tracked RESP3 connection to RESP2). A bare HELLO
// only reports connection properties and is safe.
func isProtocolChangingHelloCmd(cmd Cmder) bool {
return cmd.Name() == "hello" && len(cmd.Args()) > 1
}
// isResetCmd reports whether cmd resets all server-side connection state.
// RESET disables tracking, switches to RESP2, deauthenticates, and changes
// other state that a pooled CSC connection relies on.
func isResetCmd(cmd Cmder) bool {
return cmd.Name() == "reset"
}
// isSubscribeCmd reports whether a raw command would turn an ordinary pooled
// connection into a Pub/Sub connection. Pub/Sub pushes are deliberately left
// for the dedicated PubSub reader, so the CSC drainer cannot safely own such a
// connection.
func isSubscribeCmd(cmd Cmder) bool {
switch cmd.Name() {
case "subscribe", "psubscribe", "ssubscribe":
return true
default:
return false
}
}
// buildCacheKey returns the RESP-encoded form of the command's argument list,
// used as a collision-free canonical cache key. ok is false when the writer
// cannot marshal the arguments, in which case the caller must skip caching
// rather than bucket the command under an empty key.
func buildCacheKey(cmd Cmder) (string, bool) {
args := cmd.Args()
if len(args) == 0 {
return "", false
}
var buf bytes.Buffer
if err := proto.NewWriter(&buf).WriteArgs(args); err != nil {
return "", false
}
return buf.String(), true
}
// keyArg renders the key argument at pos exactly as proto.Writer sends it to
// the server, so invalidation lookups match the key names in the server's
// "invalidate" pushes. Only types whose stringArg rendering is byte-identical
// to the wire encoding are accepted (fmt.Sprint of any integer matches the
// writer's base-10 strconv output); for anything else — pointers, bools,
// times, durations, floats, BinaryMarshaler values — the rendering can
// diverge, the invalidation would never match, and the entry would be served
// stale forever, so ok=false and the caller skips caching (see processCached).
func keyArg(cmd Cmder, pos int) (string, bool) {
args := cmd.Args()
if pos < 0 || pos >= len(args) {
return "", false
}
switch args[pos].(type) {
case string, []byte,
int, int8, int16, int32, int64,
uint, uint8, uint16, uint32, uint64:
return cmd.stringArg(pos), true
}
return "", false
}
// extractRedisKeys returns the Redis key arguments from cmd. The result lets
// the cache map incoming invalidations back to affected entries. Returns nil
// (caller skips caching) when any key
// argument cannot be rendered in its wire form (see keyArg).
func extractRedisKeys(cmd Cmder) []string {
firstKey := cmdFirstKeyPosWithInfo(cmd, nil)
if firstKey == 0 {
return nil
}
argsLen := len(cmd.Args())
if firstKey >= argsLen {
return nil
}
switch cmd.Name() {
// All remaining args from firstKeyPos are keys.
case "mget", "exists", "sdiff", "sinter", "sunion":
keys := make([]string, 0, argsLen-firstKey)
for i := firstKey; i < argsLen; i++ {
k, ok := keyArg(cmd, i)
if !ok {
return nil
}
keys = append(keys, k)
}
return keys
// Numkeys pattern: numkeys at args[1], keys from args[2].
case "sintercard", "zdiff", "zinter", "zunion":
if argsLen < 3 {
return nil
}
numKeys, err := strconv.Atoi(cmd.stringArg(1))
if err != nil || numKeys <= 0 {
return nil
}
keys := make([]string, 0, numKeys)
for i := 2; i < 2+numKeys && i < argsLen; i++ {
k, ok := keyArg(cmd, i)
if !ok {
return nil
}
keys = append(keys, k)
}
return keys
// LCS: exactly two consecutive keys starting at firstKeyPos.
case "lcs":
if firstKey+1 >= argsLen {
return nil
}
k1, ok1 := keyArg(cmd, firstKey)
k2, ok2 := keyArg(cmd, firstKey+1)
if !ok1 || !ok2 {
return nil
}
return []string{k1, k2}
// JSON.MGET: keys from firstKeyPos to second-to-last (last arg is the
// JSON path, not a key).
case "json.mget":
lastKey := argsLen - 2
if lastKey < firstKey {
return nil
}
keys := make([]string, 0, lastKey-firstKey+1)
for i := firstKey; i <= lastKey; i++ {
k, ok := keyArg(cmd, i)
if !ok {
return nil
}
keys = append(keys, k)
}
return keys
}
// Single key at firstKeyPos (GET, HGET, LRANGE, ...).
k, ok := keyArg(cmd, firstKey)
if !ok {
return nil
}
return []string{k}
}