diff --git a/backend/_example/memory_store/go.sum b/backend/_example/memory_store/go.sum index 073f0fbb..f6a7f2a1 100644 --- a/backend/_example/memory_store/go.sum +++ b/backend/_example/memory_store/go.sum @@ -45,12 +45,12 @@ github.com/go-chi/chi v4.1.1+incompatible/go.mod h1:eB3wogJHnLi3x/kFX2A+IbTBlXxm github.com/go-chi/cors v1.1.1/go.mod h1:K2Yje0VW/SJzxiyMYu6iPQYa7hMjQX2i/F491VChg1I= github.com/go-chi/render v1.0.1 h1:4/5tis2cKaNdnv9zFLfXzcquC9HbeZgCnxGnKrltBS8= github.com/go-chi/render v1.0.1/go.mod h1:pq4Rr7HbnsdaeHagklXub+p6Wd16Af5l9koip1OvJns= -github.com/go-pkgz/auth v0.10.2/go.mod h1:w4Z1qaYvuh2P3T2gNh0f8GcKCH0HHAoQtQ8iv+9+WGg= +github.com/go-pkgz/auth v0.11.0/go.mod h1:NzVqlTW0E9JXVdAaWRq81XZjICgHnNaNdUfE3CbS2T4= github.com/go-pkgz/expirable-cache v0.0.3 h1:rTh6qNPp78z0bQE6HDhXBHUwqnV9i09Vm6dksJLXQDc= github.com/go-pkgz/expirable-cache v0.0.3/go.mod h1:+IauqN00R2FqNRLCLA+X5YljQJrwB179PfiAoMPlTlQ= github.com/go-pkgz/jrpc v0.2.0 h1:CLy/eZyekjraVrxZV18N2R1mYLMJ/nWrgdfyIOGPY/E= github.com/go-pkgz/jrpc v0.2.0/go.mod h1:wd8vtQ4CgtCnuqua6x2b1SKIgv0VSOh5Dn0uUITbiUE= -github.com/go-pkgz/lcw v0.6.1/go.mod h1:vovP88gZLeuIWn5cm0NlgPYFyGGkv3m2OcKMOOaHhj0= +github.com/go-pkgz/lcw v0.7.1/go.mod h1:3P6g9QrJsDePXEMe42ywO+tW08L17tBJGwIDdI7lZ6g= github.com/go-pkgz/lgr v0.7.0 h1:S/AAPwt/RE9a5mNJskA7dGVp+Dq6SMIW6LYjG3ITxY8= github.com/go-pkgz/lgr v0.7.0/go.mod h1:yMgxU+GobMRJgIEbSzDKy/67W18S7qmGx/7BVL5AB8Q= github.com/go-pkgz/repeater v1.1.3/go.mod h1:hVTavuO5x3Gxnu8zW7d6sQBfAneKV8X2FjU48kGfpKw= diff --git a/backend/go.mod b/backend/go.mod index 8a4f990c..8ba894dc 100644 --- a/backend/go.mod +++ b/backend/go.mod @@ -14,7 +14,7 @@ require ( github.com/go-chi/render v1.0.1 github.com/go-pkgz/auth v0.11.0 github.com/go-pkgz/jrpc v0.2.0 - github.com/go-pkgz/lcw v0.6.1 + github.com/go-pkgz/lcw v0.7.1 github.com/go-pkgz/lgr v0.7.0 github.com/go-pkgz/repeater v1.1.3 github.com/go-pkgz/rest v1.5.0 diff --git a/backend/go.sum b/backend/go.sum index a02de6f0..e292fb1a 100644 --- a/backend/go.sum +++ b/backend/go.sum @@ -62,8 +62,8 @@ github.com/go-pkgz/expirable-cache v0.0.3 h1:rTh6qNPp78z0bQE6HDhXBHUwqnV9i09Vm6d github.com/go-pkgz/expirable-cache v0.0.3/go.mod h1:+IauqN00R2FqNRLCLA+X5YljQJrwB179PfiAoMPlTlQ= github.com/go-pkgz/jrpc v0.2.0 h1:CLy/eZyekjraVrxZV18N2R1mYLMJ/nWrgdfyIOGPY/E= github.com/go-pkgz/jrpc v0.2.0/go.mod h1:wd8vtQ4CgtCnuqua6x2b1SKIgv0VSOh5Dn0uUITbiUE= -github.com/go-pkgz/lcw v0.6.1 h1:hb1v9oFaP6MiGSvROoVTiTmcyU/qHzGebUOhA74l4xU= -github.com/go-pkgz/lcw v0.6.1/go.mod h1:vovP88gZLeuIWn5cm0NlgPYFyGGkv3m2OcKMOOaHhj0= +github.com/go-pkgz/lcw v0.7.1 h1:2Q2k1am6xb2KgxZG84kL072q66rlWFUR+MvgPcDcJXA= +github.com/go-pkgz/lcw v0.7.1/go.mod h1:3P6g9QrJsDePXEMe42ywO+tW08L17tBJGwIDdI7lZ6g= github.com/go-pkgz/lgr v0.7.0 h1:S/AAPwt/RE9a5mNJskA7dGVp+Dq6SMIW6LYjG3ITxY8= github.com/go-pkgz/lgr v0.7.0/go.mod h1:yMgxU+GobMRJgIEbSzDKy/67W18S7qmGx/7BVL5AB8Q= github.com/go-pkgz/repeater v1.1.3 h1:q6+JQF14ESSy28Dd7F+wRelY4F+41HJ0LEy/szNnMiE= @@ -157,6 +157,7 @@ github.com/markbates/safe v1.0.1/go.mod h1:nAqgmRi7cY2nqMc92/bSEeQA+R4OheNU2T1kN github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU= github.com/mattn/go-colorable v0.1.4 h1:snbPLB8fVfU9iwbbo30TPtbLRzwWu6aJS6Xh4eaaviA= github.com/mattn/go-colorable v0.1.4/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE= +github.com/mattn/go-isatty v0.0.4 h1:bnP0vzxcAdeI1zdubAl5PjU6zsERjGZb7raWodagDYs= github.com/mattn/go-isatty v0.0.4/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4= github.com/mattn/go-isatty v0.0.8 h1:HLtExJ+uU2HOZ+wI0Tt5DtUDrx8yhUqDcp7fYERX4CE= github.com/mattn/go-isatty v0.0.8/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s= @@ -203,6 +204,7 @@ github.com/smartystreets/goconvey v1.6.4 h1:fv0U8FUIMPNf1L9lnHLvLhgicrIVChEkdzIK github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA= github.com/spf13/cobra v0.0.3/go.mod h1:1l0Ry5zgKvJasoi3XT1TypsSe7PqH0Sj9dhYf7v3XqQ= github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4= +github.com/stretchr/objx v0.1.0 h1:4G4v2dO3VZwixGIRoQ5Lfboy6nUhCyYzaqnIAPPhYs4= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.1.1 h1:2vfRuCMp5sSVIDSqO8oNnWJq7mPa6KVP3iPIwFBuy8A= github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= @@ -308,6 +310,7 @@ golang.org/x/sys v0.0.0-20191010194322-b09406accb47/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/sys v0.0.0-20200202164722-d101bd2416d5/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd h1:xhmwyvizuTgC2qz7ZlMluP20uW+C3Rm0FD/WLDX8884= golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs= golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk= @@ -327,6 +330,7 @@ golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8T google.golang.org/appengine v1.4.0 h1:/wp5JvzpHIxhs/dumFmF7BXTf3Z+dd4uXta4kVyO508= google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo= gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= @@ -338,6 +342,7 @@ gopkg.in/oauth2.v3 v3.12.0/go.mod h1:XEYgKqWX095YiPT+Aw5y3tCn+7/FMnlTFKrupgSiJ3I gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ= gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw= gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw= gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v2 v2.2.4 h1:/eiJrUcujPVeJ3xlSWaiNi3uSVmDGBK1pDHUHAnao1I= gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= diff --git a/backend/vendor/github.com/go-pkgz/lcw/README.md b/backend/vendor/github.com/go-pkgz/lcw/README.md index 0bdeaf6a..1ca9ffa4 100644 --- a/backend/vendor/github.com/go-pkgz/lcw/README.md +++ b/backend/vendor/github.com/go-pkgz/lcw/README.md @@ -27,8 +27,11 @@ Main features: ## Usage -``` -cache := lcw.NewLruCache(lcw.MaxKeys(500), lcw.MaxCacheSize(65536), lcw.MaxValSize(200), lcw.MaxKeySize(32)) +```go +cache, err := lcw.NewLruCache(lcw.MaxKeys(500), lcw.MaxCacheSize(65536), lcw.MaxValSize(200), lcw.MaxKeySize(32)) +if err != nil { + panic("failed to create cache") +} defer cache.Close() val, err := cache.Get("key123", func() (lcw.Value, error) { @@ -59,7 +62,7 @@ Cache can be created with URIs: 1. Key is not a string, but a composed type made from partition, key-id and list of scopes (tags). 1. Value type limited to `[]byte` 1. Added `Flush` method for scoped/tagged invalidation of multiple records in a given partition -1. A simplified interface with Get, Stat and Flush only. +1. A simplified interface with Get, Stat, Flush and Close only. ## Details diff --git a/backend/vendor/github.com/go-pkgz/lcw/cache.go b/backend/vendor/github.com/go-pkgz/lcw/cache.go index b917bd59..e7452a47 100644 --- a/backend/vendor/github.com/go-pkgz/lcw/cache.go +++ b/backend/vendor/github.com/go-pkgz/lcw/cache.go @@ -6,6 +6,8 @@ // 3 flavors of cache provided - NoP (do-nothing cache), ExpirableCache (TTL based), and LruCache package lcw +//go:generate sh -c "mockery -inpkg -name LoadingCache -print > /tmp/cache-mock.tmp && mv /tmp/cache-mock.tmp cache_mock.go" + import ( "fmt" ) @@ -81,3 +83,4 @@ func (n *Nop) Stat() CacheStat { func (n *Nop) Close() error { return nil } + diff --git a/backend/vendor/github.com/go-pkgz/lcw/eventbus/pubsub.go b/backend/vendor/github.com/go-pkgz/lcw/eventbus/pubsub.go new file mode 100644 index 00000000..3f2909f9 --- /dev/null +++ b/backend/vendor/github.com/go-pkgz/lcw/eventbus/pubsub.go @@ -0,0 +1,24 @@ +// Package eventbus provides PubSub interface used for distributed cache invalidation, +// as well as NopPubSub and RedisPubSub implementations. +package eventbus + +// PubSub interface is used for distributed cache invalidation. +// Publish is called on each entry invalidation, +// Subscribe is used for subscription for these events. +type PubSub interface { + Publish(fromID, key string) error + Subscribe(fn func(fromID, key string)) error +} + +// NopPubSub implements default do-nothing pub-sub (event bus) +type NopPubSub struct{} + +// Subscribe does nothing for NopPubSub +func (n *NopPubSub) Subscribe(fn func(fromID, key string)) error { + return nil +} + +// Publish does nothing for NopPubSub +func (n *NopPubSub) Publish(fromID, key string) error { + return nil +} diff --git a/backend/vendor/github.com/go-pkgz/lcw/eventbus/redis.go b/backend/vendor/github.com/go-pkgz/lcw/eventbus/redis.go new file mode 100644 index 00000000..851a4e9e --- /dev/null +++ b/backend/vendor/github.com/go-pkgz/lcw/eventbus/redis.go @@ -0,0 +1,72 @@ +package eventbus + +import ( + "strings" + "time" + + "github.com/go-redis/redis/v7" + "github.com/hashicorp/go-multierror" + "github.com/pkg/errors" +) + +// NewRedisPubSub creates new RedisPubSub with given parameters. +// Returns an error in case of problems with creating PubSub client for specified channel. +func NewRedisPubSub(addr, channel string) (*RedisPubSub, error) { + client := redis.NewClient(&redis.Options{Addr: addr}) + pubSub := client.Subscribe(channel) + // wait for subscription to be created and ignore the message + if _, err := pubSub.Receive(); err != nil { + _ = client.Close() + return nil, errors.Wrapf(err, "problem subscribing to channel %s on address %s", channel, addr) + } + return &RedisPubSub{client: client, pubSub: pubSub, channel: channel, done: make(chan struct{})}, nil +} + +// RedisPubSub provides Redis implementation for PubSub interface +type RedisPubSub struct { + client *redis.Client + pubSub *redis.PubSub + channel string + + done chan struct{} +} + +// Subscribe calls provided function on subscription channel provided on new RedisPubSub instance creation. +// Should not be called more than once. Spawns a goroutine and does not return an error. +func (m *RedisPubSub) Subscribe(fn func(fromID, key string)) error { + go func(done <-chan struct{}, pubsub *redis.PubSub) { + for { + select { + case <-done: + return + default: + } + msg, err := pubsub.ReceiveTimeout(time.Second * 10) + if err != nil { + continue + } + + // Process the message + if msg, ok := msg.(*redis.Message); ok { + payload := strings.Split(msg.Payload, "$") + fn(payload[0], strings.Join(payload[1:], "$")) + } + } + }(m.done, m.pubSub) + + return nil +} + +// Publish publishes provided message to channel provided on new RedisPubSub instance creation +func (m *RedisPubSub) Publish(fromID, key string) error { + return m.client.Publish(m.channel, fromID+"$"+key).Err() +} + +// Close cleans up running goroutines and closes Redis clients +func (m *RedisPubSub) Close() error { + close(m.done) + errs := new(multierror.Error) + errs = multierror.Append(errs, errors.Wrap(m.pubSub.Close(), "problem closing pubSub client")) + errs = multierror.Append(errs, errors.Wrap(m.client.Close(), "problem closing redis client")) + return errs.ErrorOrNil() +} diff --git a/backend/vendor/github.com/go-pkgz/lcw/expirable_cache.go b/backend/vendor/github.com/go-pkgz/lcw/expirable_cache.go index 4f862efd..8b20f7ec 100644 --- a/backend/vendor/github.com/go-pkgz/lcw/expirable_cache.go +++ b/backend/vendor/github.com/go-pkgz/lcw/expirable_cache.go @@ -4,8 +4,10 @@ import ( "sync/atomic" "time" + "github.com/google/uuid" "github.com/pkg/errors" + "github.com/go-pkgz/lcw/eventbus" "github.com/go-pkgz/lcw/internal/cache" ) @@ -14,6 +16,7 @@ type ExpirableCache struct { options CacheStat currentSize int64 + id string backend *cache.LoadingCache } @@ -24,7 +27,9 @@ func NewExpirableCache(opts ...Option) (*ExpirableCache, error) { maxKeys: 1000, maxValueSize: 0, ttl: 5 * time.Minute, + eventBus: &eventbus.NopPubSub{}, }, + id: uuid.New().String(), } for _, opt := range opts { @@ -33,6 +38,10 @@ func NewExpirableCache(opts ...Option) (*ExpirableCache, error) { } } + if err := res.eventBus.Subscribe(res.onBusEvent); err != nil { + return nil, errors.Wrapf(err, "can't subscribe to event bus") + } + backend, err := cache.NewLoadingCache( cache.MaxKeys(res.maxKeys), cache.TTL(res.ttl), @@ -45,6 +54,10 @@ func NewExpirableCache(opts ...Option) (*ExpirableCache, error) { size := s.Size() atomic.AddInt64(&res.currentSize, -1*int64(size)) } + // ignore the error on Publish as we don't have log inside the module and + // there is no other way to handle it: we publish the cache invalidation + // and hope for the best + _ = res.eventBus.Publish(res.id, key) }), ) if err != nil { @@ -128,6 +141,13 @@ func (c *ExpirableCache) Close() error { return nil } +// onBusEvent reacts on invalidation message triggered by event bus from another cache instance +func (c *ExpirableCache) onBusEvent(id, key string) { + if id != c.id { + c.backend.Invalidate(key) + } +} + func (c *ExpirableCache) size() int64 { return atomic.LoadInt64(&c.currentSize) } diff --git a/backend/vendor/github.com/go-pkgz/lcw/go.mod b/backend/vendor/github.com/go-pkgz/lcw/go.mod index 5ac87b08..bd7894d5 100644 --- a/backend/vendor/github.com/go-pkgz/lcw/go.mod +++ b/backend/vendor/github.com/go-pkgz/lcw/go.mod @@ -3,6 +3,7 @@ module github.com/go-pkgz/lcw require ( github.com/alicebob/miniredis/v2 v2.11.4 github.com/go-redis/redis/v7 v7.2.0 + github.com/google/uuid v1.1.1 github.com/hashicorp/go-multierror v1.1.0 github.com/hashicorp/golang-lru v0.5.4 github.com/pkg/errors v0.9.1 diff --git a/backend/vendor/github.com/go-pkgz/lcw/go.sum b/backend/vendor/github.com/go-pkgz/lcw/go.sum index 38edae2d..8f189aae 100644 --- a/backend/vendor/github.com/go-pkgz/lcw/go.sum +++ b/backend/vendor/github.com/go-pkgz/lcw/go.sum @@ -16,6 +16,8 @@ github.com/golang/protobuf v1.3.2 h1:6nsPYzhq5kReh6QImI3k5qWzO4PEbvbIW2cwSfR/6xs github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= github.com/gomodule/redigo v1.7.1-0.20190322064113-39e2c31b7ca3 h1:6amM4HsNPOvMLVc2ZnyqrjeQ92YAVWn7T4WBKK87inY= github.com/gomodule/redigo v1.7.1-0.20190322064113-39e2c31b7ca3/go.mod h1:B4C85qUVwatsJoIUNIfCRsp7qO0iAmpGFZ4EELWSbC4= +github.com/google/uuid v1.1.1 h1:Gkbcsh/GbpXz7lPftLA3P6TYMwjCLYm83jiFQZF/3gY= +github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/hashicorp/errwrap v1.0.0 h1:hLrqtEDnRye3+sgx6z4qVLNuviH3MR5aQ0ykNJa/UYA= github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= github.com/hashicorp/go-multierror v1.1.0 h1:B9UzwGQJehnUY1yNrnwREHc3fGbC2xefo8g4TbElacI= diff --git a/backend/vendor/github.com/go-pkgz/lcw/lru_cache.go b/backend/vendor/github.com/go-pkgz/lcw/lru_cache.go index fee675d7..914ab954 100644 --- a/backend/vendor/github.com/go-pkgz/lcw/lru_cache.go +++ b/backend/vendor/github.com/go-pkgz/lcw/lru_cache.go @@ -3,8 +3,11 @@ package lcw import ( "sync/atomic" + "github.com/google/uuid" lru "github.com/hashicorp/golang-lru" "github.com/pkg/errors" + + "github.com/go-pkgz/lcw/eventbus" ) // LruCache wraps lru.LruCache with loading cache Get and size limits @@ -13,6 +16,7 @@ type LruCache struct { CacheStat backend *lru.Cache currentSize int64 + id string // uuid identifying cache instance } // NewLruCache makes LRU LoadingCache implementation, 1000 max keys by default @@ -21,7 +25,9 @@ func NewLruCache(opts ...Option) (*LruCache, error) { options: options{ maxKeys: 1000, maxValueSize: 0, + eventBus: &eventbus.NopPubSub{}, }, + id: uuid.New().String(), } for _, opt := range opts { if err := opt(&res.options); err != nil { @@ -29,23 +35,33 @@ func NewLruCache(opts ...Option) (*LruCache, error) { } } + err := res.init() + return &res, err +} + +func (c *LruCache) init() error { + if err := c.eventBus.Subscribe(c.onBusEvent); err != nil { + return errors.Wrapf(err, "can't subscribe to event bus") + } + onEvicted := func(key interface{}, value interface{}) { - if res.onEvicted != nil { - res.onEvicted(key.(string), value) + if c.onEvicted != nil { + c.onEvicted(key.(string), value) } if s, ok := value.(Sizer); ok { size := s.Size() - atomic.AddInt64(&res.currentSize, -1*int64(size)) + atomic.AddInt64(&c.currentSize, -1*int64(size)) } + _ = c.eventBus.Publish(c.id, key.(string)) // signal invalidation to other nodes } var err error // OnEvicted called automatically for expired and manually deleted - if res.backend, err = lru.NewWithEvict(res.maxKeys, onEvicted); err != nil { - return nil, errors.Wrap(err, "failed to make lru cache backend") + if c.backend, err = lru.NewWithEvict(c.maxKeys, onEvicted); err != nil { + return errors.Wrap(err, "failed to make lru cache backend") } - return &res, nil + return nil } // Get gets value by key or load with fn if not found in cache @@ -131,6 +147,13 @@ func (c *LruCache) Close() error { return nil } +// onBusEvent reacts on invalidation message triggered by event bus from another cache instance +func (c *LruCache) onBusEvent(id, key string) { + if id != c.id && c.backend.Contains(key) { // prevent reaction on event from this cache + c.backend.Remove(key) + } +} + func (c *LruCache) size() int64 { return atomic.LoadInt64(&c.currentSize) } diff --git a/backend/vendor/github.com/go-pkgz/lcw/options.go b/backend/vendor/github.com/go-pkgz/lcw/options.go index 4ee964a9..5a93d2b1 100644 --- a/backend/vendor/github.com/go-pkgz/lcw/options.go +++ b/backend/vendor/github.com/go-pkgz/lcw/options.go @@ -3,6 +3,8 @@ package lcw import ( "errors" "time" + + "github.com/go-pkgz/lcw/eventbus" ) type options struct { @@ -12,6 +14,7 @@ type options struct { maxCacheSize int64 ttl time.Duration onEvicted func(key string, value Value) + eventBus eventbus.PubSub } // Option func type @@ -84,3 +87,11 @@ func OnEvicted(fn func(key string, value Value)) Option { return nil } } + +// EventBus sets PubSub for distributed cache invalidation +func EventBus(pubSub eventbus.PubSub) Option { + return func(o *options) error { + o.eventBus = pubSub + return nil + } +} diff --git a/backend/vendor/modules.txt b/backend/vendor/modules.txt index 555539ac..64dd9318 100644 --- a/backend/vendor/modules.txt +++ b/backend/vendor/modules.txt @@ -86,9 +86,10 @@ github.com/go-pkgz/expirable-cache # github.com/go-pkgz/jrpc v0.2.0 ## explicit github.com/go-pkgz/jrpc -# github.com/go-pkgz/lcw v0.6.1 +# github.com/go-pkgz/lcw v0.7.1 ## explicit github.com/go-pkgz/lcw +github.com/go-pkgz/lcw/eventbus github.com/go-pkgz/lcw/internal/cache # github.com/go-pkgz/lgr v0.7.0 ## explicit