fix(store): [OCISDEV-834] survive permanently-closed NATS connections

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

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

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"]))
+	}
+}

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
+}