Add filesystem locking on jetstream fileStore instances.

Running two instances of the server which share the same directory (e.g.
default configuration `/tmp/nats/jetstream') will corrupt each other.

We mitigate by creating an empty file called LOCK in the directory and then
acquire a posix `flock(2)` on it; a common pattern.

Signed-off-by: Jason Volk <jason@zemos.net>

server/filestore.go

@@ -174,6 +174,7 @@ type fileStore struct {
 	hh          hash.Hash64
 	qch         chan struct{}
 	cfs         []ConsumerStore
+	lfd         *os.File
 	sips        int
 	closed      bool
 	fip         bool
@@ -387,6 +388,10 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 		qch:    make(chan struct{}),
 	}
 
+	if err := fs.lockFileSystem(); err != nil {
+		return nil, err
+	}
+
 	// Set flush in place to AsyncFlush which by default is false.
 	fs.fip = !fcfg.AsyncFlush
 

server/filestore_posix.go

@@ -0,0 +1,48 @@
+// Copyright 2023 The NATS Authors
+// 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.
+
+//go:build !windows
+// +build !windows
+
+package server
+
+import (
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"syscall"
+)
+
+// Acquire filesystem-level lock for the filestore to ensure exclusive access
+// for this server instance.
+func (fs *fileStore) lockFileSystem() error {
+	var err error
+	lpath := filepath.Join(fs.fcfg.StoreDir, "LOCK")
+	fs.lfd, err = os.Create(lpath);
+	if err != nil {
+		return fmt.Errorf("could not create `%s': %v", lpath, err)
+	}
+
+	err = syscall.FcntlFlock(fs.lfd.Fd(), syscall.F_SETLK, &syscall.Flock_t{
+		Type:    syscall.F_WRLCK,
+		Whence:  io.SeekStart,
+		Start:   0,
+		Len:     0,
+	})
+	if err != nil {
+		return fmt.Errorf("lock `%s': %v", fs.lfd.Name(), err)
+	}
+
+	return nil
+}

server/filestore_windows.go

@@ -0,0 +1,51 @@
+// Copyright 2023 The NATS Authors
+// 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.
+
+//go:build windows
+// +build windows
+
+package server
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"syscall"
+	"unsafe"
+)
+
+// Acquire filesystem-level lock for the filestore to ensure exclusive access
+// for this server instance.
+func (fs *fileStore) lockFileSystem() error {
+	var (
+		err error
+		modkernel32   = syscall.NewLazyDLL("kernel32.dll")
+		prLockFileEx  = modkernel32.NewProc("LockFileEx")
+		ol            = unsafe.Pointer(new(syscall.Overlapped))
+	)
+
+	lpath := filepath.Join(fs.fcfg.StoreDir, "LOCK")
+	fs.lfd, err = os.Create(lpath);
+	if err != nil {
+		return fmt.Errorf("could not create `%s': %v", lpath, err)
+	}
+
+	a := prLockFileEx.Addr()
+	h := syscall.Handle(fs.lfd.Fd())
+	r, _, e := syscall.Syscall6(a, 6, uintptr(h), 0x3, 0, ^uintptr(0), ^uintptr(0), uintptr(ol))
+	if r == 0 {
+		return fmt.Errorf("lock `%s': %v", fs.lfd.Name(), error(e))
+	}
+
+	return nil
+}