From 8cf67d7a7a09003e3e81876d22ba9ce61fd68a32 Mon Sep 17 00:00:00 2001 From: Julian Koberg Date: Thu, 11 Jun 2026 15:50:55 +0200 Subject: [PATCH] fix(store): [OCISDEV-834] survive permanently-closed NATS connections The nats-js and nats-js-kv store backends used the NATS client defaults, which give up reconnecting after 60 attempts (~2 min). Any NATS outage longer than that left the client permanently closed, surfacing as 'nats: connection closed' on every subsequent cache operation until the process was restarted. Reconnect forever (MaxReconnect = -1) with a 5s backoff, set a proper client name, and log the previously-silent connection state transitions. Additionally, loadAttributes in the decomposedfs messagepack metadata backend no longer fails a successful disk read when the cache write-back fails: the error is logged and the disk data is returned. Co-Authored-By: Claude Opus 4.8 Signed-off-by: Julian Koberg --- .../unreleased/fix-natsjs-store-reconnect.md | 18 ++++ .../metadata/messagepack_backend.go | 6 +- .../messagepack_backend_internal_test.go | 90 +++++++++++++++++++ pkg/store/store.go | 47 +++++++--- 4 files changed, 146 insertions(+), 15 deletions(-) create mode 100644 changelog/unreleased/fix-natsjs-store-reconnect.md create mode 100644 pkg/storage/utils/decomposedfs/metadata/messagepack_backend_internal_test.go diff --git a/changelog/unreleased/fix-natsjs-store-reconnect.md b/changelog/unreleased/fix-natsjs-store-reconnect.md new file mode 100644 index 00000000000..8e59de1efff --- /dev/null +++ b/changelog/unreleased/fix-natsjs-store-reconnect.md @@ -0,0 +1,18 @@ +Bugfix: Survive permanently-closed NATS connections in the nats-js store + +The nats-js and nats-js-kv store backends used the NATS client defaults, +which give up reconnecting after 60 attempts (~2 minutes). Any NATS outage +longer than that left the client permanently closed, surfacing as +`nats: connection closed` on every subsequent cache operation until the +process was restarted. + +The store now reconnects forever (`MaxReconnect = -1`) with a 5s backoff and +logs connection state transitions (disconnect/reconnect/close), which were +previously silent. + +In addition, `loadAttributes` in the decomposedfs messagepack metadata backend +no longer fails a successful disk read when writing the attributes back into +the cache fails: the cache write-back error is logged and the data read from +disk is returned. + +https://github.com/owncloud/reva/pull/628 diff --git a/pkg/storage/utils/decomposedfs/metadata/messagepack_backend.go b/pkg/storage/utils/decomposedfs/metadata/messagepack_backend.go index 2082335f00c..2b5c5275c5c 100644 --- a/pkg/storage/utils/decomposedfs/metadata/messagepack_backend.go +++ b/pkg/storage/utils/decomposedfs/metadata/messagepack_backend.go @@ -29,6 +29,7 @@ import ( "strings" "github.com/google/renameio/v2" + "github.com/owncloud/reva/v2/pkg/appctx" "github.com/owncloud/reva/v2/pkg/storage/cache" "github.com/pkg/xattr" "github.com/rogpeppe/go-internal/lockedfile" @@ -263,7 +264,10 @@ func (b MessagePackBackend) loadAttributes(ctx context.Context, path string, sou err = b.metaCache.PushToCache(b.cacheKey(path), attribs) subspan.End() if err != nil { - return nil, err + // The attributes were read successfully from disk. A failure to + // write them back into the cache (e.g. a dead NATS connection) must + // not fail the read - log it and return the data we already have. + appctx.GetLogger(ctx).Warn().Err(err).Str("path", path).Msg("failed to write attributes to cache") } return attribs, nil diff --git a/pkg/storage/utils/decomposedfs/metadata/messagepack_backend_internal_test.go b/pkg/storage/utils/decomposedfs/metadata/messagepack_backend_internal_test.go new file mode 100644 index 00000000000..b90d236a521 --- /dev/null +++ b/pkg/storage/utils/decomposedfs/metadata/messagepack_backend_internal_test.go @@ -0,0 +1,90 @@ +// Copyright 2018-2023 CERN +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// In applying this license, CERN does not waive the privileges and immunities +// granted to it by virtue of its status as an Intergovernmental Organization +// or submit itself to any jurisdiction. + +package metadata + +import ( + "context" + "errors" + "os" + "path/filepath" + "testing" + + microstore "go-micro.dev/v4/store" + + "github.com/shamaton/msgpack/v2" +) + +// failingPushCache is a FileMetadataCache that always reports a cache miss on +// read and always fails on write, simulating a dead NATS connection backing +// the metadata cache. +type failingPushCache struct{} + +func (failingPushCache) PullFromCache(_ string, _ any) error { + return errors.New("not found") +} + +func (failingPushCache) PushToCache(_ string, _ any) error { + return errors.New("nats: connection closed") +} + +func (failingPushCache) List(_ ...microstore.ListOption) ([]string, error) { return nil, nil } +func (failingPushCache) Delete(_ string, _ ...microstore.DeleteOption) error { return nil } +func (failingPushCache) Close() error { return nil } +func (failingPushCache) RemoveMetadata(_ string) error { return nil } + +// TestLoadAttributesToleratesCacheWriteFailure verifies that a successful read +// from disk is returned even when writing the attributes back into the cache +// fails (e.g. because the NATS connection is dead). Regression test for +// OCISDEV-834. +func TestLoadAttributesToleratesCacheWriteFailure(t *testing.T) { + tmpdir, err := os.MkdirTemp(os.TempDir(), "MessagePackBackendInternalTest-") + if err != nil { + t.Fatalf("failed to create tmpdir: %v", err) + } + defer os.RemoveAll(tmpdir) + + b := MessagePackBackend{ + rootPath: filepath.Clean(tmpdir), + metaCache: failingPushCache{}, + } + + file := filepath.Join(tmpdir, "file") + want := map[string][]byte{"foo": []byte("bar")} + + // Write the metadata file directly to disk, bypassing the cache, so that + // loadAttributes has to fall through to the disk read. + d, err := msgpack.Marshal(want) + if err != nil { + t.Fatalf("failed to marshal metadata: %v", err) + } + if err := os.WriteFile(b.MetadataPath(file), d, 0600); err != nil { + t.Fatalf("failed to write metadata file: %v", err) + } + + // All() -> loadAttributes(): cache read misses, disk read succeeds, and the + // cache write-back fails. The failing PushToCache must not turn this into + // an error. + got, err := b.All(context.Background(), file) + if err != nil { + t.Fatalf("All failed despite valid on-disk metadata: %v", err) + } + if string(got["foo"]) != "bar" { + t.Fatalf("expected attribute value %q, got %q", "bar", string(got["foo"])) + } +} diff --git a/pkg/store/store.go b/pkg/store/store.go index e405a0ed15b..82365514c02 100644 --- a/pkg/store/store.go +++ b/pkg/store/store.go @@ -23,13 +23,13 @@ import ( "strings" "time" - "github.com/owncloud/reva/v2/pkg/store/etcd" - "github.com/owncloud/reva/v2/pkg/store/memory" natsjs "github.com/go-micro/plugins/v4/store/nats-js" natsjskv "github.com/go-micro/plugins/v4/store/nats-js-kv" "github.com/go-micro/plugins/v4/store/redis" redisopts "github.com/go-redis/redis/v8" "github.com/nats-io/nats.go" + "github.com/owncloud/reva/v2/pkg/store/etcd" + "github.com/owncloud/reva/v2/pkg/store/memory" "go-micro.dev/v4/logger" microstore "go-micro.dev/v4/store" ) @@ -125,12 +125,7 @@ func Create(opts ...microstore.Option) microstore.Store { // TODO nats needs a DefaultTTL option as it does not support per Write TTL ... // FIXME nats has restrictions on the key, we cannot use slashes AFAICT // host, port, clusterid - natsOptions := nats.GetDefaultOptions() - natsOptions.Name = "TODO" // we can pass in the service name to allow identifying the client, but that requires adding a custom context option - if auth, ok := options.Context.Value(authenticationContextKey{}).([]string); ok && len(auth) == 2 { - natsOptions.User = auth[0] - natsOptions.Password = auth[1] - } + natsOptions := defaultNatsOptions(options) return natsjs.NewStore( append(opts, natsjs.NatsOptions(natsOptions), // always pass in properly initialized default nats options @@ -143,12 +138,7 @@ func Create(opts ...microstore.Option) microstore.Store { opts = append(opts, natsjskv.DefaultMemory()) } - natsOptions := nats.GetDefaultOptions() - natsOptions.Name = "TODO" // we can pass in the service name to allow identifying the client, but that requires adding a custom context option - if auth, ok := options.Context.Value(authenticationContextKey{}).([]string); ok && len(auth) == 2 { - natsOptions.User = auth[0] - natsOptions.Password = auth[1] - } + natsOptions := defaultNatsOptions(options) return natsjskv.NewStore( append(opts, natsjskv.NatsOptions(natsOptions), // always pass in properly initialized default nats options @@ -166,3 +156,32 @@ func Create(opts ...microstore.Option) microstore.Store { return microstore.NewMemoryStore(opts...) } } + +// defaultNatsOptions builds the nats.Options shared by the nats-js and +// nats-js-kv store backends. It overrides the NATS client defaults so the +// client never permanently gives up on a closed connection: the default +// MaxReconnect (60) combined with the default ReconnectWait (2s) means any +// NATS outage longer than ~2 minutes leaves the client permanently closed, +// which the store plugins then surface as "nats: connection closed" on every +// subsequent operation. Reconnecting forever, together with the connection +// state handlers, keeps the client alive and makes the transitions visible. +func defaultNatsOptions(options *microstore.Options) nats.Options { + natsOptions := nats.GetDefaultOptions() + natsOptions.Name = "reva-store" + natsOptions.MaxReconnect = -1 // reconnect forever; the default of 60 gives up after ~2 minutes + natsOptions.ReconnectWait = 5 * time.Second + if auth, ok := options.Context.Value(authenticationContextKey{}).([]string); ok && len(auth) == 2 { + natsOptions.User = auth[0] + natsOptions.Password = auth[1] + } + natsOptions.DisconnectedErrCB = func(_ *nats.Conn, err error) { + logger.Logf(logger.WarnLevel, "reva-store: nats connection disconnected: %v", err) + } + natsOptions.ReconnectedCB = func(c *nats.Conn) { + logger.Logf(logger.InfoLevel, "reva-store: nats connection reconnected to %s", c.ConnectedUrl()) + } + natsOptions.ClosedCB = func(_ *nats.Conn) { + logger.Logf(logger.ErrorLevel, "reva-store: nats connection closed") + } + return natsOptions +}