amadan

carlos / platform Public

Coordinate SQLite VFS connections and retain failed cleanup ownership

claude 36aa460296f6532f1611b46d458a98038eb0596a parent a7eea08
8 files changed, +1532 −166
  • internal/vfsspike/db_locks_linux.go +199 −0
  • internal/vfsspike/lock_failure_linux.go +55 −0
  • internal/vfsspike/lock_failure_linux_test.go +266 −0
  • internal/vfsspike/locks_linux.go +123 −129
  • internal/vfsspike/multiconnection_linux_test.go +206 −0
  • internal/vfsspike/process_linux_test.go +567 −0
  • internal/vfsspike/vfs_linux.go +75 −23
  • internal/vfsspike/vfs_linux_test.go +41 −14

CI passed — run details

diff --git a/internal/vfsspike/db_locks_linux.go b/internal/vfsspike/db_locks_linux.go
new file mode 100644
index 000000000..30c4b5b5f
--- /dev/null
+++ b/internal/vfsspike/db_locks_linux.go
@@ -0,0 +1,199 @@
+//go:build linux && (amd64 || arm64)
+
+package vfsspike
+
+import (
+ "errors"
+ "golang.org/x/sys/unix"
+ "modernc.org/libc"
+ sqlite "modernc.org/sqlite/lib"
+)
+
+const pending int64 = 0x40000000
+
+type dbOwnership struct {
+ readers int
+ reserved, pending, exclusive *file
+ deferred []int
+}
+
+func setLock(fd int, kind int16, start, length int64) error {
+ return unix.FcntlFlock(uintptr(fd), unix.F_SETLK, &unix.Flock_t{Type: kind, Whence: 0, Start: start, Len: length})
+}
+func lockResult(err error) int32 {
+ if err == nil {
+ return sqlite.SQLITE_OK
+ }
+ if errors.Is(err, unix.EAGAIN) || errors.Is(err, unix.EACCES) {
+ return sqlite.SQLITE_BUSY
+ }
+ return sqlite.SQLITE_IOERR_LOCK
+}
+func lockFile(t *libc.TLS, p uintptr, want int32) int32 {
+ f := getFile(p)
+ f.v.mu.Lock()
+ defer f.v.mu.Unlock()
+ if f.v.poisoned.Load() {
+ return sqlite.SQLITE_IOERR_LOCK
+ }
+ n := &f.v.db
+ if !f.main {
+ return sqlite.SQLITE_IOERR_LOCK
+ }
+ if want <= f.lock {
+ return sqlite.SQLITE_OK
+ }
+ if f.lock == sqlite.SQLITE_LOCK_NONE {
+ if n.pending != nil || n.exclusive != nil {
+ return sqlite.SQLITE_BUSY
+ }
+ if n.readers == 0 {
+ if rc := lockResult(f.v.setLock(f.fd, unix.F_RDLCK, pending, 1)); rc != 0 {
+ return rc
+ }
+ err := f.v.setLock(f.fd, unix.F_RDLCK, pending+2, 510)
+ unerr := f.v.setLock(f.fd, unix.F_UNLCK, pending, 1)
+ if err != nil {
+ return lockResult(err)
+ }
+ if unerr != nil {
+ return sqlite.SQLITE_IOERR_UNLOCK
+ }
+ }
+ n.readers++
+ f.lock = sqlite.SQLITE_LOCK_SHARED
+ }
+ if want == sqlite.SQLITE_LOCK_SHARED {
+ return sqlite.SQLITE_OK
+ }
+ if want == sqlite.SQLITE_LOCK_RESERVED {
+ if (n.reserved != nil && n.reserved != f) || (n.pending != nil && n.pending != f) {
+ return sqlite.SQLITE_BUSY
+ }
+ if rc := lockResult(f.v.setLock(f.fd, unix.F_WRLCK, pending+1, 1)); rc != 0 {
+ return rc
+ }
+ n.reserved = f
+ f.lock = want
+ return sqlite.SQLITE_OK
+ }
+ if want != sqlite.SQLITE_LOCK_EXCLUSIVE && want != sqlite.SQLITE_LOCK_PENDING {
+ return sqlite.SQLITE_IOERR_LOCK
+ }
+ if (n.reserved != nil && n.reserved != f) || (n.pending != nil && n.pending != f) {
+ return sqlite.SQLITE_BUSY
+ }
+ if n.pending != f {
+ if rc := lockResult(f.v.setLock(f.fd, unix.F_WRLCK, pending, 1)); rc != 0 {
+ return rc
+ }
+ n.pending = f
+ f.lock = sqlite.SQLITE_LOCK_PENDING
+ }
+ if want == sqlite.SQLITE_LOCK_PENDING {
+ return sqlite.SQLITE_OK
+ }
+ if n.readers > 1 {
+ return sqlite.SQLITE_BUSY
+ }
+ if rc := lockResult(f.v.setLock(f.fd, unix.F_WRLCK, pending+2, 510)); rc != 0 {
+ return rc
+ }
+ n.exclusive = f
+ f.lock = want
+ return sqlite.SQLITE_OK
+}
+func unlockFile(t *libc.TLS, p uintptr, want int32) int32 {
+ f := getFile(p)
+ f.v.mu.Lock()
+ defer f.v.mu.Unlock()
+ return unlockDB(f, want)
+}
+func unlockDB(f *file, want int32) (rc int32) {
+ if f.v.poisoned.Load() {
+ return sqlite.SQLITE_IOERR_UNLOCK
+ }
+ defer func() {
+ if rc != sqlite.SQLITE_OK {
+ f.v.poisoned.Store(true)
+ }
+ }()
+ n := &f.v.db
+ if want != sqlite.SQLITE_LOCK_NONE && want != sqlite.SQLITE_LOCK_SHARED {
+ return sqlite.SQLITE_IOERR_UNLOCK
+ }
+ if want >= f.lock {
+ return sqlite.SQLITE_OK
+ }
+ if n.exclusive == f {
+ kind := int16(unix.F_RDLCK)
+ if want == sqlite.SQLITE_LOCK_NONE {
+ kind = unix.F_UNLCK
+ }
+ if f.v.setLock(f.fd, kind, pending+2, 510) != nil {
+ return sqlite.SQLITE_IOERR_UNLOCK
+ }
+ n.exclusive = nil
+ } else if want == sqlite.SQLITE_LOCK_NONE && n.readers == 1 {
+ if f.v.setLock(f.fd, unix.F_UNLCK, pending+2, 510) != nil {
+ return sqlite.SQLITE_IOERR_UNLOCK
+ }
+ }
+ if n.pending == f {
+ if f.v.setLock(f.fd, unix.F_UNLCK, pending, 1) != nil {
+ return sqlite.SQLITE_IOERR_UNLOCK
+ }
+ n.pending = nil
+ }
+ if n.reserved == f {
+ if f.v.setLock(f.fd, unix.F_UNLCK, pending+1, 1) != nil {
+ return sqlite.SQLITE_IOERR_UNLOCK
+ }
+ n.reserved = nil
+ }
+ if want == sqlite.SQLITE_LOCK_NONE {
+ n.readers--
+ }
+ f.lock = want
+ if n.readers == 0 {
+ for _, fd := range n.deferred {
+ unix.Close(fd)
+ }
+ n.deferred = nil
+ }
+ return sqlite.SQLITE_OK
+}
+func reservedFile(t *libc.TLS, p, out uintptr) int32 {
+ f := getFile(p)
+ f.v.mu.Lock()
+ defer f.v.mu.Unlock()
+ foreignStore(t, out, int32(0))
+ if f.v.poisoned.Load() {
+ return sqlite.SQLITE_IOERR_CHECKRESERVEDLOCK
+ }
+ if f.v.db.reserved != nil || f.v.db.pending != nil || f.v.db.exclusive != nil {
+ foreignStore(t, out, int32(1))
+ return sqlite.SQLITE_OK
+ }
+ l := unix.Flock_t{Type: unix.F_WRLCK, Whence: 0, Start: pending + 1, Len: 1}
+ if unix.FcntlFlock(uintptr(f.fd), unix.F_GETLK, &l) != nil {
+ return sqlite.SQLITE_IOERR_CHECKRESERVEDLOCK
+ }
+ if l.Type != unix.F_UNLCK {
+ foreignStore(t, out, int32(1))
+ }
+ return sqlite.SQLITE_OK
+}
+
+func (v *VFS) setLock(fd int, kind int16, start, length int64) error {
+ var err error
+ if v.lockHook != nil {
+ err = v.lockHook(fd, kind, start, length)
+ } else {
+ err = setLock(fd, kind, start, length)
+ }
+ if err != nil && (kind == unix.F_UNLCK || (!errors.Is(err, unix.EAGAIN) && !errors.Is(err, unix.EACCES))) {
+ v.poisoned.Store(true)
+ }
+ return err
+}
diff --git a/internal/vfsspike/lock_failure_linux.go b/internal/vfsspike/lock_failure_linux.go
new file mode 100644
index 000000000..012b60523
--- /dev/null
+++ b/internal/vfsspike/lock_failure_linux.go
@@ -0,0 +1,55 @@
+//go:build linux && (amd64 || arm64)
+
+package vfsspike
+
+import (
+ "golang.org/x/sys/unix"
+ sqlite "modernc.org/sqlite/lib"
+)
+
+func (v *VFS) finishPoisoned() int32 {
+ if len(v.files) != 0 {
+ return sqlite.SQLITE_OK
+ }
+ rc := int32(sqlite.SQLITE_OK)
+ if node := v.shm; node != nil {
+ for page, b := range node.regions {
+ if unix.Munmap(b) != nil {
+ rc = sqlite.SQLITE_IOERR_SHMMAP
+ } else {
+ delete(node.regions, page)
+ }
+ }
+ if node.fd >= 0 {
+ if unix.Close(node.fd) != nil {
+ rc = sqlite.SQLITE_IOERR_CLOSE
+ }
+ node.fd = -1
+ }
+ if len(node.regions) == 0 {
+ v.shm = nil
+ }
+ }
+ for _, f := range v.retired {
+ if f.fd >= 0 {
+ if unix.Close(f.fd) != nil {
+ rc = sqlite.SQLITE_IOERR_CLOSE
+ }
+ f.fd = -1
+ }
+ }
+ v.retired = nil
+ for _, fd := range v.db.deferred {
+ if unix.Close(fd) != nil {
+ rc = sqlite.SQLITE_IOERR_CLOSE
+ }
+ }
+ v.db = dbOwnership{}
+ for _, fd := range v.rejected {
+ if unix.Close(fd) != nil {
+ rc = sqlite.SQLITE_IOERR_CLOSE
+ }
+ }
+ v.rejected = nil
+ return rc
+}
diff --git a/internal/vfsspike/lock_failure_linux_test.go b/internal/vfsspike/lock_failure_linux_test.go
new file mode 100644
index 000000000..732485beb
--- /dev/null
+++ b/internal/vfsspike/lock_failure_linux_test.go
@@ -0,0 +1,266 @@
+//go:build linux && (amd64 || arm64)
+
+package vfsspike
+
+import (
+ "fmt"
+ "golang.org/x/sys/unix"
+ sqlite "modernc.org/sqlite/lib"
+ "path/filepath"
+ "testing"
+)
+
+func injectLockFailure(v *VFS, step int) {
+ calls := 0
+ v.lockHook = func(fd int, kind int16, start, length int64) error {
+ calls++
+ if calls == step {
+ return unix.EIO
+ }
+ return setLock(fd, kind, start, length)
+ }
+}
+func requireLock(t *testing.T, v *VFS, p uintptr, want int32) {
+ t.Helper()
+ if rc := lockFile(v.tls, p, want); rc != sqlite.SQLITE_OK {
+ t.Fatalf("lock %d: %d", want, rc)
+ }
+}
+func TestDBUnlockFailurePoisonsVFS(t *testing.T) {
+ for _, want := range []int32{sqlite.SQLITE_LOCK_SHARED, sqlite.SQLITE_LOCK_NONE} {
+ for step := 1; step <= 3; step++ {
+ t.Run(fmt.Sprintf("target%d-step%d", want, step), func(t *testing.T) {
+ v, dir := setup(t)
+ a := openMainCallback(t, v)
+ b := openMainCallback(t, v)
+ requireLock(t, v, a, sqlite.SQLITE_LOCK_SHARED)
+ requireLock(t, v, a, sqlite.SQLITE_LOCK_RESERVED)
+ requireLock(t, v, a, sqlite.SQLITE_LOCK_EXCLUSIVE)
+ injectLockFailure(v, step)
+ rc := unlockFile(v.tls, a, want)
+ v.lockHook = nil
+ if rc != sqlite.SQLITE_IOERR_UNLOCK {
+ t.Fatalf("unlock %d", rc)
+ }
+ expected := int16(unix.F_WRLCK)
+ if step > 1 {
+ expected = unix.F_RDLCK
+ if want == sqlite.SQLITE_LOCK_NONE {
+ expected = unix.F_UNLCK
+ }
+ }
+ assertKernelLock(t, filepath.Join(dir, "main.db"), pending+2, 510, expected)
+ if rc := lockFile(v.tls, a, sqlite.SQLITE_LOCK_EXCLUSIVE); rc == sqlite.SQLITE_OK {
+ t.Error("failed unlock allowed stale exclusive success")
+ }
+ if rc := lockFile(v.tls, b, sqlite.SQLITE_LOCK_SHARED); rc == sqlite.SQLITE_OK {
+ t.Error("failed unlock allowed another connection")
+ }
+ })
+ }
+ }
+}
+func TestFinalDBCloseFailureReleasesKernelLock(t *testing.T) {
+ v, dir := setup(t)
+ a := openMainCallback(t, v)
+ fd := getFile(a).fd
+ requireLock(t, v, a, sqlite.SQLITE_LOCK_SHARED)
+ injectLockFailure(v, 1)
+ rc := closeFile(v.tls, a)
+ v.lockHook = nil
+ if rc != sqlite.SQLITE_IOERR_UNLOCK {
+ t.Fatalf("close %d", rc)
+ }
+ assertKernelLock(t, filepath.Join(dir, "main.db"), pending+2, 510, unix.F_UNLCK)
+ if _, err := unix.FcntlInt(uintptr(fd), unix.F_GETFD, 0); err != unix.EBADF {
+ t.Errorf("failed final close retained fd: %v", err)
+ }
+}
+func TestFinalSHMCloseFailureReleasesKernelLock(t *testing.T) {
+ v, dir := setup(t)
+ a := openMainCallback(t, v)
+ out := sqliteBuffer(t, v.tls, 8)
+ if rc := shmMap(v.tls, a, 0, 32768, 1, out); rc != sqlite.SQLITE_OK {
+ t.Fatal(rc)
+ }
+ if rc := shmLock(v.tls, a, 0, 1, sqlite.SQLITE_SHM_LOCK|sqlite.SQLITE_SHM_EXCLUSIVE); rc != sqlite.SQLITE_OK {
+ t.Fatal(rc)
+ }
+ fd := v.shm.fd
+ injectLockFailure(v, 1)
+ rc := closeFile(v.tls, a)
+ v.lockHook = nil
+ if rc != sqlite.SQLITE_IOERR_SHMLOCK {
+ t.Fatalf("close %d", rc)
+ }
+ assertKernelLock(t, filepath.Join(dir, "main.db-shm"), 120, 1, unix.F_UNLCK)
+ if _, err := unix.FcntlInt(uintptr(fd), unix.F_GETFD, 0); err != unix.EBADF {
+ t.Errorf("failed final SHM close retained fd: %v", err)
+ }
+ if v.shm != nil {
+ t.Error("failed final SHM close retained users/mappings")
+ }
+}
+
+func TestPoisonedCloseRetainsOwnershipUntilFinalFile(t *testing.T) {
+ for _, shm := range []bool{false, true} {
+ t.Run(fmt.Sprint(shm), func(t *testing.T) {
+ v, dir := setup(t)
+ a := openMainCallback(t, v)
+ b := openMainCallback(t, v)
+ afd := getFile(a).fd
+ path := filepath.Join(dir, "main.db")
+ start := pending + 2
+ length := int64(510)
+ if shm {
+ out := sqliteBuffer(t, v.tls, 8)
+ if rc := shmMap(v.tls, a, 0, 32768, 1, out); rc != 0 {
+ t.Fatal(rc)
+ }
+ if rc := shmMap(v.tls, b, 0, 32768, 1, out); rc != 0 {
+ t.Fatal(rc)
+ }
+ if rc := shmLock(v.tls, a, 0, 1, sqlite.SQLITE_SHM_LOCK|sqlite.SQLITE_SHM_EXCLUSIVE); rc != 0 {
+ t.Fatal(rc)
+ }
+ path += "-shm"
+ start = 120
+ length = 1
+ } else {
+ requireLock(t, v, a, sqlite.SQLITE_LOCK_SHARED)
+ requireLock(t, v, a, sqlite.SQLITE_LOCK_RESERVED)
+ requireLock(t, v, a, sqlite.SQLITE_LOCK_EXCLUSIVE)
+ }
+ injectLockFailure(v, 1)
+ rc := closeFile(v.tls, a)
+ v.lockHook = nil
+ if rc == sqlite.SQLITE_OK {
+ t.Fatal("failed close succeeded")
+ }
+ if getFile(a) != nil || len(v.retired) != 1 {
+ t.Fatal("closed callback was not replaced by internal ownership")
+ }
+ if _, err := unix.FcntlInt(uintptr(afd), unix.F_GETFD, 0); err != nil {
+ t.Fatal("retired descriptor closed before last custom file")
+ }
+ assertKernelLock(t, path, start, length, unix.F_WRLCK)
+ if err := v.Close(); err != unix.EBUSY {
+ t.Fatalf("unregister with live file: %v", err)
+ }
+ if rc := closeFile(v.tls, b); rc != sqlite.SQLITE_OK {
+ t.Fatal(rc)
+ }
+ assertKernelLock(t, path, start, length, unix.F_UNLCK)
+ if len(v.retired) != 0 || len(v.db.deferred) != 0 || v.shm != nil {
+ t.Fatal("final cleanup retained ownership")
+ }
+ if _, err := unix.FcntlInt(uintptr(afd), unix.F_GETFD, 0); err != unix.EBADF {
+ t.Fatalf("retired descriptor survived final cleanup: %v", err)
+ }
+ })
+ }
+}
+func TestSHMPartialUnlockFailurePoisonsVFS(t *testing.T) {
+ for step := 1; step <= 3; step++ {
+ t.Run(fmt.Sprint(step), func(t *testing.T) {
+ v, dir := setup(t)
+ a := openMainCallback(t, v)
+ b := openMainCallback(t, v)
+ out := sqliteBuffer(t, v.tls, 8)
+ if rc := shmMap(v.tls, a, 0, 32768, 1, out); rc != 0 {
+ t.Fatal(rc)
+ }
+ if rc := shmMap(v.tls, b, 0, 32768, 1, out); rc != 0 {
+ t.Fatal(rc)
+ }
+ if rc := shmLock(v.tls, a, 0, 3, sqlite.SQLITE_SHM_LOCK|sqlite.SQLITE_SHM_EXCLUSIVE); rc != 0 {
+ t.Fatal(rc)
+ }
+ injectLockFailure(v, step)
+ rc := shmLock(v.tls, a, 0, 3, sqlite.SQLITE_SHM_UNLOCK|sqlite.SQLITE_SHM_EXCLUSIVE)
+ v.lockHook = nil
+ if rc != sqlite.SQLITE_IOERR_SHMLOCK {
+ t.Fatalf("unlock %d", rc)
+ }
+ for slot := 0; slot < 3; slot++ {
+ want := int16(unix.F_WRLCK)
+ if slot < step-1 {
+ want = unix.F_UNLCK
+ }
+ assertKernelLock(t, filepath.Join(dir, "main.db-shm"), 120+int64(slot), 1, want)
+ }
+ if rc := shmLock(v.tls, a, 0, 3, sqlite.SQLITE_SHM_LOCK|sqlite.SQLITE_SHM_EXCLUSIVE); rc == 0 {
+ t.Error("poisoned SHM owner reacquired")
+ }
+ if rc := shmLock(v.tls, b, 0, 1, sqlite.SQLITE_SHM_LOCK|sqlite.SQLITE_SHM_SHARED); rc == 0 {
+ t.Error("poisoned SHM admitted another connection")
+ }
+ })
+ }
+}
+func TestAcquisitionFailurePoisonsIO(t *testing.T) {
+ for _, tc := range []struct {
+ level int32
+ step int
+ }{{sqlite.SQLITE_LOCK_SHARED, 1}, {sqlite.SQLITE_LOCK_SHARED, 2}, {sqlite.SQLITE_LOCK_SHARED, 3}, {sqlite.SQLITE_LOCK_RESERVED, 1}, {sqlite.SQLITE_LOCK_EXCLUSIVE, 1}, {sqlite.SQLITE_LOCK_EXCLUSIVE, 2}} {
+ t.Run(fmt.Sprintf("level%d-step%d", tc.level, tc.step), func(t *testing.T) {
+ v, _ := setup(t)
+ a := openMainCallback(t, v)
+ if tc.level >= sqlite.SQLITE_LOCK_RESERVED {
+ requireLock(t, v, a, sqlite.SQLITE_LOCK_SHARED)
+ }
+ if tc.level == sqlite.SQLITE_LOCK_EXCLUSIVE {
+ requireLock(t, v, a, sqlite.SQLITE_LOCK_RESERVED)
+ }
+ injectLockFailure(v, tc.step)
+ rc := lockFile(v.tls, a, tc.level)
+ v.lockHook = nil
+ if rc == sqlite.SQLITE_OK || rc == sqlite.SQLITE_BUSY {
+ t.Fatalf("permanent failure code %d", rc)
+ }
+ if !v.poisoned.Load() {
+ t.Fatal("permanent failure not latched")
+ }
+ out := sqliteBuffer(t, v.tls, 16)
+ for name, call := range map[string]func() int32{
+ "read": func() int32 { return readFile(v.tls, a, out, 1, 0) }, "write": func() int32 { return writeFile(v.tls, a, out, 1, 0) }, "truncate": func() int32 { return truncateFile(v.tls, a, 0) }, "sync": func() int32 { return syncFile(v.tls, a, 0) }, "size": func() int32 { return sizeFile(v.tls, a, out) }, "control": func() int32 { return controlFile(v.tls, a, sqlite.SQLITE_FCNTL_LOCKSTATE, out) }, "lock": func() int32 { return lockFile(v.tls, a, tc.level) }, "reserved": func() int32 { return reservedFile(v.tls, a, out) }, "shmmap": func() int32 { return shmMap(v.tls, a, 0, 32768, 1, out) },
+ } {
+ if got := call(); got == sqlite.SQLITE_OK {
+ t.Errorf("poison allowed %s", name)
+ }
+ }
+ if got := openFile(v.tls, v.ptr, 0, out, sqlite.SQLITE_OPEN_TEMP_DB|sqlite.SQLITE_OPEN_READWRITE|sqlite.SQLITE_OPEN_CREATE, 0); got == sqlite.SQLITE_OK {
+ t.Error("poison allowed open")
+ }
+ })
+ }
+}
+
+func TestSQLiteConnectionsCloseAfterLockFailure(t *testing.T) {
+ v, dir := setup(t)
+ a := connect(t, v)
+ execSQL(t, a, "CREATE TABLE data(value); INSERT INTO data VALUES (1)")
+ b := connect(t, v)
+ scalar(t, b, "SELECT count(*) FROM data")
+ execSQL(t, a, "BEGIN EXCLUSIVE; INSERT INTO data VALUES (2)")
+ v.lockHook = func(fd int, kind int16, start, length int64) error {
+ if kind == unix.F_UNLCK && start == pending+1 {
+ return unix.EIO
+ }
+ return setLock(fd, kind, start, length)
+ }
+ _, err := a.Exec("COMMIT")
+ v.lockHook = nil
+ if err == nil {
+ t.Fatal("commit did not report unlock failure")
+ }
+ if _, err := b.Exec("INSERT INTO data VALUES (3)"); err == nil {
+ t.Fatal("second connection wrote after poison")
+ }
+ a.Close()
+ b.Close()
+ if len(v.files) != 0 || len(v.retired) != 0 || len(v.db.deferred) != 0 || v.shm != nil {
+ t.Fatal("SQLite close retained ownership")
+ }
+ assertKernelLock(t, filepath.Join(dir, "main.db"), pending, 512, unix.F_UNLCK)
+}
diff --git a/internal/vfsspike/locks_linux.go b/internal/vfsspike/locks_linux.go
index 5d1282499..761aebd1c 100644
--- a/internal/vfsspike/locks_linux.go
+++ b/internal/vfsspike/locks_linux.go
@@ -10,116 +10,20 @@ import (
"unsafe"
)
-const pending int64 = 0x40000000
-
-func setLock(fd int, kind int16, start, length int64) error {
- return unix.FcntlFlock(uintptr(fd), unix.F_SETLK, &unix.Flock_t{Type: kind, Whence: 0, Start: start, Len: length})
-}
-func lockResult(err error) int32 {
- if err == nil {
- return sqlite.SQLITE_OK
- }
- if errors.Is(err, unix.EAGAIN) || errors.Is(err, unix.EACCES) {
- return sqlite.SQLITE_BUSY
- }
- return sqlite.SQLITE_IOERR_LOCK
-}
-func lockFile(t *libc.TLS, p uintptr, want int32) int32 {
- f := getFile(p)
- f.v.mu.Lock()
- defer f.v.mu.Unlock()
- if !f.main {
- return sqlite.SQLITE_IOERR_LOCK
- }
- if want <= f.lock {
- return sqlite.SQLITE_OK
- }
- if f.lock == sqlite.SQLITE_LOCK_NONE {
- if rc := lockResult(setLock(f.fd, unix.F_RDLCK, pending, 1)); rc != 0 {
- return rc
- }
- err := setLock(f.fd, unix.F_RDLCK, pending+2, 510)
- unlockErr := setLock(f.fd, unix.F_UNLCK, pending, 1)
- if err != nil {
- return lockResult(err)
- }
- if unlockErr != nil {
- return sqlite.SQLITE_IOERR_UNLOCK
- }
- f.lock = sqlite.SQLITE_LOCK_SHARED
- }
- if want == sqlite.SQLITE_LOCK_SHARED {
- return sqlite.SQLITE_OK
- }
- if want == sqlite.SQLITE_LOCK_RESERVED {
- if rc := lockResult(setLock(f.fd, unix.F_WRLCK, pending+1, 1)); rc != 0 {
- return rc
- }
- f.lock = want
- return sqlite.SQLITE_OK
- }
- if want != sqlite.SQLITE_LOCK_EXCLUSIVE && want != sqlite.SQLITE_LOCK_PENDING {
- return sqlite.SQLITE_IOERR_LOCK
- }
- if f.lock < sqlite.SQLITE_LOCK_PENDING {
- if rc := lockResult(setLock(f.fd, unix.F_WRLCK, pending, 1)); rc != 0 {
- return rc
- }
- f.lock = sqlite.SQLITE_LOCK_PENDING
- }
- if want == sqlite.SQLITE_LOCK_PENDING {
- return sqlite.SQLITE_OK
- }
- if rc := lockResult(setLock(f.fd, unix.F_WRLCK, pending+2, 510)); rc != 0 {
- return rc
- }
- f.lock = want
- return sqlite.SQLITE_OK
-}
-func unlockFile(t *libc.TLS, p uintptr, want int32) int32 {
- f := getFile(p)
- f.v.mu.Lock()
- defer f.v.mu.Unlock()
- if want != sqlite.SQLITE_LOCK_NONE && want != sqlite.SQLITE_LOCK_SHARED {
- return sqlite.SQLITE_IOERR_UNLOCK
- }
- if want >= f.lock {
- return sqlite.SQLITE_OK
- }
- if want == sqlite.SQLITE_LOCK_NONE {
- if setLock(f.fd, unix.F_UNLCK, pending, 512) != nil {
- return sqlite.SQLITE_IOERR_UNLOCK
- }
- } else {
- if f.lock == sqlite.SQLITE_LOCK_EXCLUSIVE && setLock(f.fd, unix.F_RDLCK, pending+2, 510) != nil {
- return sqlite.SQLITE_IOERR_UNLOCK
- }
- if setLock(f.fd, unix.F_UNLCK, pending, 2) != nil {
- return sqlite.SQLITE_IOERR_UNLOCK
- }
- }
- f.lock = want
- return sqlite.SQLITE_OK
-}
-func reservedFile(t *libc.TLS, p, out uintptr) int32 {
- f := getFile(p)
- f.v.mu.Lock()
- defer f.v.mu.Unlock()
- foreignStore(t, out, int32(0))
- if f.lock >= sqlite.SQLITE_LOCK_RESERVED {
- foreignStore(t, out, int32(1))
- return sqlite.SQLITE_OK
- }
- l := unix.Flock_t{Type: unix.F_WRLCK, Whence: 0, Start: pending + 1, Len: 1}
- if unix.FcntlFlock(uintptr(f.fd), unix.F_GETLK, &l) != nil {
- return sqlite.SQLITE_IOERR_CHECKRESERVEDLOCK
- }
- if l.Type != unix.F_UNLCK {
- foreignStore(t, out, int32(1))
- }
- return sqlite.SQLITE_OK
+type sharedMemory struct {
+ fd int
+ id identity
+ users map[*file]bool
+ regions map[int32][]byte
+ shared [8]int
+ exclusive [8]*file
+ persistent bool
}
+
func ensureSHM(f *file) int32 {
+ if f.v.poisoned.Load() {
+ return sqlite.SQLITE_IOERR_SHMOPEN
+ }
if f.shm >= 0 {
return sqlite.SQLITE_OK
}
@@ -127,6 +31,18 @@ func ensureSHM(f *file) int32 {
return sqlite.SQLITE_IOERR_SHMOPEN
}
name := f.name + "-shm"
+ if node := f.v.shm; node != nil {
+ var s unix.Stat_t
+ if unix.Fstatat(f.v.data, name, &s, unix.AT_SYMLINK_NOFOLLOW) != nil || !same(&s, node.id) || s.Nlink != 1 {
+ return sqlite.SQLITE_IOERR_SHMOPEN
+ }
+ node.users[f] = true
+ f.shm = node.fd
+ f.shmID = node.id
+ f.regions = node.regions
+ return sqlite.SQLITE_OK
+ }
+
var before unix.Stat_t
if err := unix.Fstatat(f.v.data, name, &before, unix.AT_SYMLINK_NOFOLLOW); err == nil && (before.Mode&unix.S_IFMT != unix.S_IFREG || same(&before, f.v.expected)) {
return sqlite.SQLITE_IOERR_SHMOPEN
@@ -140,7 +56,7 @@ func ensureSHM(f *file) int32 {
f.v.reject(fd)
return sqlite.SQLITE_IOERR_SHMOPEN
}
- if err = setLock(fd, unix.F_WRLCK, 128, 1); err == nil {
+ if err = f.v.setLock(fd, unix.F_WRLCK, 128, 1); err == nil {
if unix.Ftruncate(fd, 0) != nil {
f.v.reject(fd)
return sqlite.SQLITE_IOERR_SHMSIZE
@@ -149,12 +65,13 @@ func ensureSHM(f *file) int32 {
f.v.reject(fd)
return sqlite.SQLITE_IOERR_SHMLOCK
}
- if setLock(fd, unix.F_RDLCK, 128, 1) != nil {
+ if err := f.v.setLock(fd, unix.F_RDLCK, 128, 1); err != nil {
f.v.reject(fd)
- return sqlite.SQLITE_BUSY
+ return shmResult(lockResult(err))
}
f.shm = fd
f.shmID = identity{s.Dev, s.Ino}
+ f.v.shm = &sharedMemory{fd: fd, id: f.shmID, users: map[*file]bool{f: true}, regions: f.regions}
return sqlite.SQLITE_OK
}
func shmMap(t *libc.TLS, p uintptr, page, size, extend int32, out uintptr) int32 {
@@ -206,45 +123,122 @@ func shmLock(t *libc.TLS, p uintptr, offset, n, flags int32) int32 {
if rc := ensureSHM(f); rc != 0 {
return rc
}
- var kind int16
+ node := f.v.shm
switch flags {
case sqlite.SQLITE_SHM_LOCK | sqlite.SQLITE_SHM_SHARED:
- kind = unix.F_RDLCK
+ if n != 1 {
+ return sqlite.SQLITE_IOERR_SHMLOCK
+ }
+ bit := uint16(1) << offset
+ if node.exclusive[offset] != nil {
+ return sqlite.SQLITE_BUSY
+ }
+ if f.shared&bit != 0 {
+ return sqlite.SQLITE_OK
+ }
+ if node.shared[offset] == 0 {
+ if rc := lockResult(f.v.setLock(node.fd, unix.F_RDLCK, 120+int64(offset), 1)); rc != 0 {
+ return shmResult(rc)
+ }
+ }
+ node.shared[offset]++
+ f.shared |= bit
case sqlite.SQLITE_SHM_LOCK | sqlite.SQLITE_SHM_EXCLUSIVE:
- kind = unix.F_WRLCK
+ for i := offset; i < offset+n; i++ {
+ if node.shared[i] != 0 || (node.exclusive[i] != nil && node.exclusive[i] != f) {
+ return sqlite.SQLITE_BUSY
+ }
+ }
+ if rc := lockResult(f.v.setLock(node.fd, unix.F_WRLCK, 120+int64(offset), int64(n))); rc != 0 {
+ return shmResult(rc)
+ }
+ for i := offset; i < offset+n; i++ {
+ node.exclusive[i] = f
+ f.exclusive |= uint16(1) << i
+ }
case sqlite.SQLITE_SHM_UNLOCK | sqlite.SQLITE_SHM_SHARED, sqlite.SQLITE_SHM_UNLOCK | sqlite.SQLITE_SHM_EXCLUSIVE:
- kind = unix.F_UNLCK
+ return unlockSHM(f, offset, n)
default:
return sqlite.SQLITE_IOERR_SHMLOCK
}
- if rc := lockResult(setLock(f.shm, kind, 120+int64(offset), int64(n))); rc != 0 {
- if rc == sqlite.SQLITE_BUSY {
- return rc
+ return sqlite.SQLITE_OK
+}
+func shmResult(rc int32) int32 {
+ if rc == sqlite.SQLITE_BUSY {
+ return rc
+ }
+ return sqlite.SQLITE_IOERR_SHMLOCK
+}
+func unlockSHM(f *file, offset, n int32) int32 {
+ node := f.v.shm
+ for i := offset; i < offset+n; i++ {
+ bit := uint16(1) << i
+ if f.exclusive&bit != 0 {
+ if f.v.setLock(node.fd, unix.F_UNLCK, 120+int64(i), 1) != nil {
+ return sqlite.SQLITE_IOERR_SHMLOCK
+ }
+ node.exclusive[i] = nil
+ f.exclusive &^= bit
+ }
+ if f.shared&bit != 0 {
+ if node.shared[i] == 1 {
+ if f.v.setLock(node.fd, unix.F_UNLCK, 120+int64(i), 1) != nil {
+ return sqlite.SQLITE_IOERR_SHMLOCK
+ }
+ }
+ node.shared[i]--
+ f.shared &^= bit
}
- return sqlite.SQLITE_IOERR_SHMLOCK
}
return sqlite.SQLITE_OK
}
func shmBarrier(t *libc.TLS, p uintptr) { barrier.Add(1) }
-func unmapLocked(f *file, remove bool) int32 {
- rc := int32(sqlite.SQLITE_OK)
- for page, b := range f.regions {
+func unmapLocked(f *file, remove bool) (rc int32) {
+ if f.v.poisoned.Load() {
+ return sqlite.SQLITE_IOERR_SHMLOCK
+ }
+ defer func() {
+ if rc != sqlite.SQLITE_OK {
+ f.v.poisoned.Store(true)
+ }
+ }()
+ if f.shm < 0 {
+ return sqlite.SQLITE_OK
+ }
+ node := f.v.shm
+ if rc := unlockSHM(f, 0, 8); rc != sqlite.SQLITE_OK {
+ return rc
+ }
+ node.persistent = node.persistent || f.persistent
+ delete(node.users, f)
+ f.shm = -1
+ f.regions = make(map[int32][]byte)
+ if len(node.users) > 0 {
+ return sqlite.SQLITE_OK
+ }
+ rc = int32(sqlite.SQLITE_OK)
+ for page, b := range node.regions {
if unix.Munmap(b) != nil {
rc = sqlite.SQLITE_IOERR_SHMMAP
+ } else {
+ delete(node.regions, page)
}
- delete(f.regions, page)
}
- if f.shm >= 0 {
- if remove && !f.persistent && f.v.recheck() == nil && setLock(f.shm, unix.F_WRLCK, 128, 1) == nil {
+ if rc != sqlite.SQLITE_OK {
+ return rc
+ }
+ if remove && !node.persistent && f.v.recheck() == nil && f.v.setLock(node.fd, unix.F_WRLCK, 128, 1) == nil {
+ var s unix.Stat_t
+ if unix.Fstatat(f.v.data, f.name+"-shm", &s, unix.AT_SYMLINK_NOFOLLOW) == nil && same(&s, node.id) {
if err := unix.Unlinkat(f.v.data, f.name+"-shm", 0); err != nil && !errors.Is(err, unix.ENOENT) {
rc = sqlite.SQLITE_IOERR_DELETE
}
}
- if unix.Close(f.shm) != nil {
- rc = sqlite.SQLITE_IOERR_CLOSE
- }
- f.shm = -1
}
+ if unix.Close(node.fd) != nil {
+ rc = sqlite.SQLITE_IOERR_CLOSE
+ }
+ f.v.shm = nil
return rc
}
func shmUnmap(t *libc.TLS, p uintptr, remove int32) int32 {
diff --git a/internal/vfsspike/multiconnection_linux_test.go b/internal/vfsspike/multiconnection_linux_test.go
new file mode 100644
index 000000000..de1e3e6d4
--- /dev/null
+++ b/internal/vfsspike/multiconnection_linux_test.go
@@ -0,0 +1,206 @@
+//go:build linux && (amd64 || arm64)
+
+package vfsspike
+
+import (
+ "database/sql"
+ "golang.org/x/sys/unix"
+ "path/filepath"
+ "testing"
+)
+
+func scalar(t *testing.T, db *sql.DB, q string) int {
+ t.Helper()
+ var n int
+ if err := db.QueryRow(q).Scan(&n); err != nil {
+ t.Fatal(err)
+ }
+ return n
+}
+func busySQL(t *testing.T, db *sql.DB, q string) {
+ t.Helper()
+ _, err := db.Exec(q)
+ coded, ok := err.(interface{ Code() int })
+ if !ok || coded.Code()&255 != 5 {
+ t.Fatalf("%s: expected SQLITE_BUSY, got %v", q, err)
+ }
+}
+func onlyMain(t *testing.T, v *VFS) *file {
+ t.Helper()
+ var found *file
+ for _, f := range v.files {
+ if f.main {
+ if found != nil {
+ t.Fatal("multiple main files")
+ }
+ found = f
+ }
+ }
+ if found == nil {
+ t.Fatal("missing main")
+ }
+ return found
+}
+func TestDBSharedCountAndDeferredClose(t *testing.T) {
+ v, dir := setup(t)
+ a := connect(t, v)
+ execSQL(t, a, "CREATE TABLE data(value); INSERT INTO data VALUES (1)")
+ af := onlyMain(t, v)
+ b := connect(t, v)
+ execSQL(t, a, "BEGIN")
+ execSQL(t, b, "BEGIN")
+ scalar(t, a, "SELECT count(*) FROM data")
+ scalar(t, b, "SELECT count(*) FROM data")
+ path := filepath.Join(dir, "main.db")
+ assertKernelLock(t, path, pending+2, 510, unix.F_RDLCK)
+ execSQL(t, a, "ROLLBACK")
+ assertKernelLock(t, path, pending+2, 510, unix.F_RDLCK)
+ if err := a.Close(); err != nil {
+ t.Fatal(err)
+ }
+ assertKernelLock(t, path, pending+2, 510, unix.F_RDLCK)
+ if _, err := unix.FcntlInt(uintptr(af.fd), unix.F_GETFD, 0); err != nil {
+ t.Fatalf("descriptor closed under surviving reader: %v", err)
+ }
+ if scalar(t, b, "SELECT count(*) FROM data") != 1 {
+ t.Fatal("reader changed")
+ }
+ execSQL(t, b, "ROLLBACK")
+ assertKernelLock(t, path, pending+2, 510, unix.F_UNLCK)
+ if _, err := unix.FcntlInt(uintptr(af.fd), unix.F_GETFD, 0); err != unix.EBADF {
+ t.Fatalf("deferred descriptor not drained: %v", err)
+ }
+}
+func TestDBWriterLocalExclusionAndPending(t *testing.T) {
+ v, dir := setup(t)
+ a := connect(t, v)
+ execSQL(t, a, "CREATE TABLE data(value); INSERT INTO data VALUES (1)")
+ b := connect(t, v)
+ execSQL(t, b, "BEGIN")
+ scalar(t, b, "SELECT count(*) FROM data")
+ execSQL(t, a, "BEGIN IMMEDIATE; INSERT INTO data VALUES (2)")
+ busySQL(t, b, "INSERT INTO data VALUES (3)")
+ busySQL(t, a, "COMMIT")
+ c := connect(t, v)
+ busySQL(t, c, "SELECT * FROM data")
+ assertReserved(t, filepath.Join(dir, "main.db"))
+ execSQL(t, b, "ROLLBACK")
+ execSQL(t, a, "COMMIT")
+ if scalar(t, c, "SELECT count(*) FROM data") != 2 {
+ t.Fatal("committed row missing")
+ }
+}
+func TestWALWriterExclusionAndOtherClose(t *testing.T) {
+ v, dir := setup(t)
+ a := connect(t, v)
+ execSQL(t, a, "PRAGMA journal_mode=WAL; CREATE TABLE data(value); INSERT INTO data VALUES (1)")
+ b := connect(t, v)
+ scalar(t, b, "SELECT count(*) FROM data")
+ execSQL(t, a, "BEGIN IMMEDIATE; INSERT INTO data VALUES (2)")
+ busySQL(t, b, "BEGIN IMMEDIATE")
+ path := filepath.Join(dir, "main.db-shm")
+ assertKernelLock(t, path, 120, 1, unix.F_WRLCK)
+ if err := b.Close(); err != nil {
+ t.Fatal(err)
+ }
+ assertKernelLock(t, path, 120, 1, unix.F_WRLCK)
+ execSQL(t, a, "COMMIT")
+ assertKernelLock(t, path, 120, 1, unix.F_UNLCK)
+ execSQL(t, a, "PRAGMA wal_checkpoint(TRUNCATE)")
+ if scalar(t, a, "SELECT count(*) FROM data") != 2 {
+ t.Fatal("writer lost data")
+ }
+}
+func TestWALSharedCountsAndReaderSurvivesClose(t *testing.T) {
+ v, dir := setup(t)
+ a := connect(t, v)
+ execSQL(t, a, "PRAGMA journal_mode=WAL; CREATE TABLE data(value); INSERT INTO data VALUES (1)")
+ af := onlyMain(t, v)
+ b := connect(t, v)
+ execSQL(t, a, "BEGIN")
+ execSQL(t, b, "BEGIN")
+ scalar(t, a, "SELECT count(*) FROM data")
+ scalar(t, b, "SELECT count(*) FROM data")
+ var bf *file
+ for _, f := range v.files {
+ if f.main && f != af {
+ bf = f
+ }
+ }
+ if bf == nil {
+ t.Fatal("second main missing")
+ }
+ common := af.shared & bf.shared
+ slot := -1
+ for i := 3; i < 8; i++ {
+ if common&(1<<i) != 0 {
+ slot = i
+ break
+ }
+ }
+ if slot < 0 {
+ t.Fatalf("no common SHM read slot: %x %x", af.shared, bf.shared)
+ }
+ path := filepath.Join(dir, "main.db-shm")
+ assertKernelLock(t, path, 120+int64(slot), 1, unix.F_RDLCK)
+ execSQL(t, a, "ROLLBACK")
+ assertKernelLock(t, path, 120+int64(slot), 1, unix.F_RDLCK)
+ if err := a.Close(); err != nil {
+ t.Fatal(err)
+ }
+ assertKernelLock(t, path, 120+int64(slot), 1, unix.F_RDLCK)
+ if scalar(t, b, "SELECT count(*) FROM data") != 1 {
+ t.Fatal("reader lost snapshot")
+ }
+ execSQL(t, b, "ROLLBACK")
+ assertKernelLock(t, path, 120+int64(slot), 1, unix.F_UNLCK)
+ execSQL(t, b, "INSERT INTO data VALUES (2); PRAGMA wal_checkpoint(TRUNCATE)")
+}
+
+func TestConcurrentCustomWriterExclusion(t *testing.T) {
+ for _, mode := range []string{"DELETE", "WAL"} {
+ t.Run(mode, func(t *testing.T) {
+ v, _ := setup(t)
+ a := connect(t, v)
+ execSQL(t, a, "PRAGMA journal_mode="+mode+"; CREATE TABLE data(value)")
+ b := connect(t, v)
+ scalar(t, b, "SELECT count(*) FROM data")
+ for i := 0; i < 20; i++ {
+ start := make(chan struct{})
+ release := make(chan struct{})
+ results := make(chan error, 2)
+ done := make(chan error, 2)
+ for _, db := range []*sql.DB{a, b} {
+ go func(db *sql.DB) {
+ <-start
+ _, err := db.Exec("BEGIN IMMEDIATE")
+ results <- err
+ <-release
+ if err == nil {
+ _, err = db.Exec("ROLLBACK")
+ }
+ done <- err
+ }(db)
+ }
+ close(start)
+ success := 0
+ var bad error
+ for j := 0; j < 2; j++ {
+ err := <-results
+ if err == nil {
+ success++
+ } else if coded, ok := err.(interface{ Code() int }); !ok || coded.Code()&255 != 5 {
+ bad = err
+ }
+ }
+ close(release)
+ for j := 0; j < 2; j++ {
+ <-done
+ }
+ if success != 1 || bad != nil {
+ t.Fatalf("iteration %d: writers %d error %v", i, success, bad)
+ }
+ }
+ })
+ }
+}
diff --git a/internal/vfsspike/process_linux_test.go b/internal/vfsspike/process_linux_test.go
new file mode 100644
index 000000000..7de94a420
--- /dev/null
+++ b/internal/vfsspike/process_linux_test.go
@@ -0,0 +1,567 @@
+//go:build linux && (amd64 || arm64)
+
+package vfsspike
+
+import (
+ "context"
+ "database/sql"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "reflect"
+ "strings"
+ "testing"
+ "time"
+
+ "golang.org/x/sys/unix"
+)
+
+type processRequest struct{ Op, SQL string }
+type processResponse struct {
+ Error string
+ Code int
+ Rows [][]int64
+ Inodes []uint64
+ TempFiles []string
+}
+
+func TestVFSProcessChild(t *testing.T) {
+ if os.Getenv("VFSSPIKE_PROCESS_CHILD") != "1" {
+ t.Skip("child only")
+ }
+ socket := os.NewFile(3, "rights")
+ defer socket.Close()
+ payload, control := make([]byte, 1), make([]byte, unix.CmsgSpace(8))
+ n, oobn, flags, _, err := unix.Recvmsg(3, payload, control, unix.MSG_CMSG_CLOEXEC)
+ if err != nil || n != 1 || flags&unix.MSG_CTRUNC != 0 {
+ t.Fatalf("receive: %d %d %v", n, flags, err)
+ }
+ messages, err := unix.ParseSocketControlMessage(control[:oobn])
+ if err != nil {
+ t.Fatal(err)
+ }
+ var fds []int
+ for _, message := range messages {
+ got, err := unix.ParseUnixRights(&message)
+ if err != nil {
+ t.Fatal(err)
+ }
+ fds = append(fds, got...)
+ }
+ if len(fds) != 2 {
+ t.Fatalf("descriptor count %d", len(fds))
+ }
+ data, meta := os.NewFile(uintptr(fds[0]), "DATA"), os.NewFile(uintptr(fds[1]), "META")
+ var ids []uint64
+ for _, f := range []*os.File{data, meta} {
+ var st unix.Stat_t
+ if err := unix.Fstat(int(f.Fd()), &st); err != nil {
+ t.Fatal(err)
+ }
+ if st.Mode&unix.S_IFMT != unix.S_IFDIR {
+ t.Fatal("not a directory")
+ }
+ ids = append(ids, st.Dev, st.Ino)
+ }
+ v, err := Register(data, meta, "main.db")
+ data.Close()
+ meta.Close()
+ if err != nil {
+ t.Fatal(err)
+ }
+ enc, dec := json.NewEncoder(os.Stdout), json.NewDecoder(os.Stdin)
+ if err := enc.Encode(processResponse{Inodes: ids}); err != nil {
+ t.Fatal(err)
+ }
+ var db *sql.DB
+ for {
+ var req processRequest
+ if err := dec.Decode(&req); err != nil {
+ if !errors.Is(err, io.EOF) {
+ t.Fatal(err)
+ }
+ break
+ }
+ response := processResponse{}
+ var err error
+ switch req.Op {
+ case "open":
+ db, err = sql.Open("sqlite", "file:main.db?vfs="+v.Name)
+ if err == nil {
+ db.SetMaxOpenConns(1)
+ _, err = db.Exec("PRAGMA busy_timeout=0; PRAGMA wal_autocheckpoint=0")
+ }
+ case "close":
+ if db != nil {
+ err = db.Close()
+ db = nil
+ }
+ case "exec":
+ _, err = db.Exec(req.SQL)
+ case "temp":
+ _, err = db.Exec("PRAGMA temp_store=FILE; PRAGMA temp.cache_size=1; CREATE TEMP TABLE spill(payload BLOB); WITH RECURSIVE n(v) AS (VALUES(1) UNION ALL SELECT v+1 FROM n WHERE v<1000) INSERT INTO spill SELECT zeroblob(8192) FROM n")
+ if err == nil {
+ v.mu.Lock()
+ for _, f := range v.files {
+ if f.temporary {
+ name, e := os.Readlink(fmt.Sprintf("/proc/self/fd/%d", f.fd))
+ if e != nil {
+ err = e
+ break
+ }
+ response.TempFiles = append(response.TempFiles, name)
+ }
+ }
+ v.mu.Unlock()
+ }
+ case "query":
+ var rows *sql.Rows
+ rows, err = db.Query(req.SQL)
+ if err == nil {
+ columns, e := rows.Columns()
+ err = e
+ for err == nil && rows.Next() {
+ row := make([]int64, len(columns))
+ ptrs := make([]any, len(row))
+ for i := range row {
+ ptrs[i] = &row[i]
+ }
+ err = rows.Scan(ptrs...)
+ if err == nil {
+ response.Rows = append(response.Rows, row)
+ }
+ }
+ if err == nil {
+ err = rows.Err()
+ }
+ rows.Close()
+ }
+ case "exit":
+ if db != nil {
+ err = db.Close()
+ db = nil
+ }
+ if err == nil {
+ err = v.Close()
+ }
+ default:
+ err = fmt.Errorf("unknown test operation %q", req.Op)
+ }
+ if err != nil {
+ response.Error = err.Error()
+ var coded interface{ Code() int }
+ if errors.As(err, &coded) {
+ response.Code = coded.Code()
+ }
+ }
+ if err := enc.Encode(response); err != nil {
+ t.Fatal(err)
+ }
+ if req.Op == "exit" {
+ os.Exit(0)
+ }
+ }
+ if db != nil {
+ db.Close()
+ }
+ if err := v.Close(); err != nil {
+ t.Fatal(err)
+ }
+}
+
+type vfsProcess struct {
+ t *testing.T
+ cmd *exec.Cmd
+ enc *json.Encoder
+ replies chan processResponse
+ done chan error
+ exited bool
+}
+
+func launchVFSProcess(t *testing.T, data, meta string) *vfsProcess {
+ t.Helper()
+ pair, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_SEQPACKET|unix.SOCK_CLOEXEC, 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ parent, child := os.NewFile(uintptr(pair[0]), "parent-rights"), os.NewFile(uintptr(pair[1]), "child-rights")
+ defer parent.Close()
+ defer child.Close()
+ exe, err := os.Executable()
+ if err != nil {
+ t.Fatal(err)
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second)
+ args := []string{"-test.run=^TestVFSProcessChild$"}
+ if runner := os.Getenv("VFSSPIKE_TEST_RUNNER"); runner != "" {
+ args = append([]string{exe}, args...)
+ exe = runner
+ }
+ cmd := exec.CommandContext(ctx, exe, args...)
+ cmd.Env = append(os.Environ(), "VFSSPIKE_PROCESS_CHILD=1")
+ cmd.ExtraFiles = []*os.File{child}
+ cmd.Stderr = os.Stderr
+ input, err := cmd.StdinPipe()
+ if err != nil {
+ cancel()
+ t.Fatal(err)
+ }
+ output, writer, err := os.Pipe()
+ if err != nil {
+ cancel()
+ t.Fatal(err)
+ }
+ cmd.Stdout = writer
+ if err := cmd.Start(); err != nil {
+ output.Close()
+ writer.Close()
+ input.Close()
+ cancel()
+ t.Fatal(err)
+ }
+ writer.Close()
+ p := &vfsProcess{t: t, cmd: cmd, enc: json.NewEncoder(input), replies: make(chan processResponse, 1), done: make(chan error, 1)}
+ go func() {
+ defer output.Close()
+ dec := json.NewDecoder(output)
+ for {
+ var r processResponse
+ if err := dec.Decode(&r); err != nil {
+ close(p.replies)
+ return
+ }
+ p.replies <- r
+ }
+ }()
+ go func() { p.done <- cmd.Wait() }()
+ t.Cleanup(func() {
+ input.Close()
+ if !p.exited {
+ cmd.Process.Kill()
+ <-p.done
+ p.exited = true
+ }
+ cancel()
+ })
+ df, err := os.Open(data)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer df.Close()
+ mf, err := os.Open(meta)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer mf.Close()
+ if err := unix.Sendmsg(int(parent.Fd()), []byte{1}, unix.UnixRights(int(df.Fd()), int(mf.Fd())), nil, 0); err != nil {
+ t.Fatal(err)
+ }
+ ready := p.receive()
+ var expected []uint64
+ for _, f := range []*os.File{df, mf} {
+ var st unix.Stat_t
+ if err := unix.Fstat(int(f.Fd()), &st); err != nil {
+ t.Fatal(err)
+ }
+ expected = append(expected, st.Dev, st.Ino)
+ }
+ if !reflect.DeepEqual(ready.Inodes, expected) {
+ t.Fatalf("SCM_RIGHTS identities got %v want %v", ready.Inodes, expected)
+ }
+ t.Logf("SCM_RIGHTS child pid=%d DATA=(%d,%d) META=(%d,%d)", cmd.Process.Pid, expected[0], expected[1], expected[2], expected[3])
+ return p
+}
+func (p *vfsProcess) receive() processResponse {
+ p.t.Helper()
+ select {
+ case r, ok := <-p.replies:
+ if !ok {
+ p.t.Fatal("child response stream ended")
+ }
+ return r
+ case <-time.After(10 * time.Second):
+ p.t.Fatal("child response timed out")
+ return processResponse{}
+ }
+}
+func (p *vfsProcess) call(op, q string) processResponse {
+ p.t.Helper()
+ if err := p.enc.Encode(processRequest{op, q}); err != nil {
+ p.t.Fatal(err)
+ }
+ return p.receive()
+}
+func (p *vfsProcess) ok(op, q string) processResponse {
+ p.t.Helper()
+ r := p.call(op, q)
+ if r.Error != "" {
+ p.t.Fatalf("child %s %s: %s", op, q, r.Error)
+ }
+ return r
+}
+func (p *vfsProcess) busy(q string) {
+ p.t.Helper()
+ r := p.call("exec", q)
+ if r.Code&255 != 5 {
+ p.t.Fatalf("child wanted BUSY for %s: %+v", q, r)
+ }
+}
+func (p *vfsProcess) stop(crash bool) {
+ p.t.Helper()
+ if crash {
+ if err := p.cmd.Process.Kill(); err != nil {
+ p.t.Fatal(err)
+ }
+ } else {
+ p.ok("exit", "")
+ }
+ select {
+ case err := <-p.done:
+ p.exited = true
+ if !crash && err != nil {
+ p.t.Fatal(err)
+ }
+ case <-time.After(10 * time.Second):
+ p.t.Fatal("child exit timed out")
+ }
+}
+func (p *vfsProcess) count(want int64) {
+ p.t.Helper()
+ r := p.ok("query", "SELECT count(*) FROM data")
+ if !reflect.DeepEqual(r.Rows, [][]int64{{want}}) {
+ p.t.Fatalf("child count %+v want %d", r, want)
+ }
+}
+func processOrdinary(t *testing.T, path string) *sql.DB {
+ t.Helper()
+ db, err := sql.Open("sqlite", path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ db.SetMaxOpenConns(1)
+ t.Cleanup(func() { db.Close() })
+ execSQL(t, db, "PRAGMA busy_timeout=0; PRAGMA wal_autocheckpoint=0")
+ return db
+}
+func ordinaryBusy(t *testing.T, db *sql.DB, q string) {
+ t.Helper()
+ _, err := db.Exec(q)
+ var coded interface{ Code() int }
+ if !errors.As(err, &coded) || coded.Code()&255 != 5 {
+ t.Fatalf("ordinary wanted BUSY for %s: %v", q, err)
+ }
+}
+func ordinaryCount(t *testing.T, db *sql.DB, want int64) {
+ t.Helper()
+ var n int64
+ if err := db.QueryRow("SELECT count(*) FROM data").Scan(&n); err != nil || n != want {
+ t.Fatalf("ordinary count %d want %d: %v", n, want, err)
+ }
+}
+func processDirs(t *testing.T) (string, string) {
+ t.Helper()
+ root := t.TempDir()
+ data, meta := filepath.Join(root, "data"), filepath.Join(root, "meta")
+ for _, dir := range []string{data, meta} {
+ if err := os.Mkdir(dir, 0700); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if err := os.WriteFile(filepath.Join(data, "main.db"), nil, 0600); err != nil {
+ t.Fatal(err)
+ }
+ return data, meta
+}
+
+func TestVFSProcessDescriptorRenameAndReopen(t *testing.T) {
+ data, meta := processDirs(t)
+ p := launchVFSProcess(t, data, meta)
+ for _, dir := range []string{data, meta} {
+ if err := os.Rename(dir, dir+"-moved"); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Mkdir(dir, 0700); err != nil {
+ t.Fatal(err)
+ }
+ }
+ p.ok("open", "")
+ p.ok("exec", "PRAGMA journal_mode=WAL; CREATE TABLE data(value INTEGER); INSERT INTO data VALUES(1)")
+ temp := p.ok("temp", "")
+ if len(temp.TempFiles) == 0 {
+ t.Fatal("SQLite did not spill a temporary file")
+ }
+ for _, name := range temp.TempFiles {
+ if !strings.HasPrefix(name, meta+"-moved/") {
+ t.Fatalf("temporary file outside received META: %q", name)
+ }
+ }
+ t.Logf("SQLite temporary files through renamed META descriptor: %v", temp.TempFiles)
+ db := processOrdinary(t, filepath.Join(data+"-moved", "main.db"))
+ ordinaryCount(t, db, 1)
+ execSQL(t, db, "INSERT INTO data VALUES(2)")
+ p.count(2)
+ p.ok("query", "PRAGMA wal_checkpoint(TRUNCATE)")
+ p.ok("close", "")
+ p.ok("open", "")
+ p.count(2)
+ p.stop(false)
+ for _, dir := range []string{data, meta} {
+ entries, err := os.ReadDir(dir)
+ if err != nil || len(entries) != 0 {
+ t.Fatalf("replacement directory touched %s: %v %v", dir, entries, err)
+ }
+ }
+}
+
+func TestVFSProcessWriterExclusionAndClose(t *testing.T) {
+ for _, mode := range []string{"DELETE", "WAL"} {
+ t.Run(mode, func(t *testing.T) {
+ data, meta := processDirs(t)
+ db := processOrdinary(t, filepath.Join(data, "main.db"))
+ execSQL(t, db, "PRAGMA journal_mode="+mode+"; CREATE TABLE data(value INTEGER)")
+ p := launchVFSProcess(t, data, meta)
+ p.ok("open", "")
+ execSQL(t, db, "BEGIN IMMEDIATE")
+ p.busy("BEGIN IMMEDIATE")
+ execSQL(t, db, "ROLLBACK")
+ p.ok("exec", "BEGIN IMMEDIATE")
+ ordinaryBusy(t, db, "BEGIN IMMEDIATE")
+ p.ok("exec", "ROLLBACK")
+ execSQL(t, db, "BEGIN IMMEDIATE")
+ p.ok("close", "")
+ p.ok("open", "")
+ p.busy("BEGIN IMMEDIATE")
+ execSQL(t, db, "ROLLBACK")
+ p.ok("exec", "BEGIN IMMEDIATE")
+ p.ok("close", "")
+ execSQL(t, db, "BEGIN IMMEDIATE; ROLLBACK")
+ p.stop(false)
+ })
+ }
+}
+func TestVFSProcessRollbackReaderBlocksCommit(t *testing.T) {
+ data, meta := processDirs(t)
+ db := processOrdinary(t, filepath.Join(data, "main.db"))
+ execSQL(t, db, "CREATE TABLE data(value INTEGER); INSERT INTO data VALUES(1)")
+ p := launchVFSProcess(t, data, meta)
+ p.ok("open", "")
+ execSQL(t, db, "BEGIN")
+ ordinaryCount(t, db, 1)
+ p.ok("exec", "BEGIN IMMEDIATE; INSERT INTO data VALUES(2)")
+ p.busy("COMMIT")
+ ordinaryCount(t, db, 1)
+ execSQL(t, db, "ROLLBACK")
+ p.ok("exec", "COMMIT")
+ ordinaryCount(t, db, 2)
+ p.stop(false)
+}
+func TestVFSProcessWALSnapshotsAndCheckpoints(t *testing.T) {
+ data, meta := processDirs(t)
+ db := processOrdinary(t, filepath.Join(data, "main.db"))
+ execSQL(t, db, "PRAGMA journal_mode=WAL; CREATE TABLE data(value INTEGER); INSERT INTO data VALUES(1)")
+ p := launchVFSProcess(t, data, meta)
+ p.ok("open", "")
+ execSQL(t, db, "BEGIN")
+ ordinaryCount(t, db, 1)
+ p.ok("exec", "INSERT INTO data VALUES(2)")
+ ordinaryCount(t, db, 1)
+ r := p.ok("query", "PRAGMA wal_checkpoint(TRUNCATE)")
+ if len(r.Rows) != 1 || r.Rows[0][0] != 1 {
+ t.Fatalf("checkpoint did not report blocked reader: %+v", r)
+ }
+ execSQL(t, db, "ROLLBACK")
+ r = p.ok("query", "PRAGMA wal_checkpoint(TRUNCATE)")
+ if r.Rows[0][0] != 0 {
+ t.Fatalf("checkpoint remained blocked: %+v", r)
+ }
+ p.ok("exec", "BEGIN")
+ p.count(2)
+ execSQL(t, db, "INSERT INTO data VALUES(3)")
+ p.count(2)
+ var busy, log, checkpointed int64
+ if err := db.QueryRow("PRAGMA wal_checkpoint(TRUNCATE)").Scan(&busy, &log, &checkpointed); err != nil || busy != 1 {
+ t.Fatalf("ordinary checkpoint busy=%d: %v", busy, err)
+ }
+ p.ok("exec", "ROLLBACK")
+ if err := db.QueryRow("PRAGMA wal_checkpoint(TRUNCATE)").Scan(&busy, &log, &checkpointed); err != nil || busy != 0 {
+ t.Fatalf("ordinary checkpoint busy=%d: %v", busy, err)
+ }
+ p.count(3)
+ p.stop(false)
+}
+func TestVFSProcessCrashRecovery(t *testing.T) {
+ for _, mode := range []string{"DELETE", "WAL"} {
+ t.Run(mode, func(t *testing.T) {
+ data, meta := processDirs(t)
+ db := processOrdinary(t, filepath.Join(data, "main.db"))
+ execSQL(t, db, "PRAGMA journal_mode="+mode+"; CREATE TABLE data(value INTEGER,payload BLOB)")
+ p := launchVFSProcess(t, data, meta)
+ p.ok("open", "")
+ p.ok("exec", "INSERT INTO data VALUES(1,zeroblob(8192))")
+ spillPath := filepath.Join(data, "main.db")
+ if mode == "WAL" {
+ spillPath += "-wal"
+ }
+ before, err := os.Stat(spillPath)
+ if err != nil {
+ t.Fatal(err)
+ }
+ p.ok("exec", "PRAGMA cache_size=1; BEGIN IMMEDIATE; WITH RECURSIVE n(v) AS (VALUES(2) UNION ALL SELECT v+1 FROM n WHERE v<100) INSERT INTO data SELECT v,zeroblob(8192) FROM n")
+ after, err := os.Stat(spillPath)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if after.Size() <= before.Size() {
+ t.Fatalf("uncommitted transaction did not spill: before=%d after=%d", before.Size(), after.Size())
+ }
+ t.Logf("uncommitted %s spill grew %s from %d to %d bytes before SIGKILL", mode, filepath.Base(spillPath), before.Size(), after.Size())
+ p.stop(true)
+ ordinaryCount(t, db, 1)
+ var check string
+ if err := db.QueryRow("PRAGMA integrity_check").Scan(&check); err != nil || check != "ok" {
+ t.Fatalf("integrity %q: %v", check, err)
+ }
+ execSQL(t, db, "INSERT INTO data VALUES(2,NULL)")
+ ordinaryCount(t, db, 2)
+ next := launchVFSProcess(t, data, meta)
+ next.ok("open", "")
+ next.count(2)
+ next.stop(false)
+ })
+ }
+}
+func TestVFSProcessSameProcessNegativeControl(t *testing.T) {
+ v, dir := setup(t)
+ custom := connect(t, v)
+ execSQL(t, custom, "CREATE TABLE data(value INTEGER)")
+ ordinary := processOrdinary(t, filepath.Join(dir, "main.db"))
+ execSQL(t, ordinary, "BEGIN IMMEDIATE")
+ if _, err := custom.Exec("BEGIN IMMEDIATE"); err != nil {
+ t.Fatalf("negative control did not reproduce overlapping process-owned locks: %v", err)
+ }
+ t.Log("ordinary and custom connections simultaneously hold BEGIN IMMEDIATE in one process; helper isolation is required")
+ execSQL(t, custom, "ROLLBACK")
+ execSQL(t, ordinary, "ROLLBACK")
+}
+
+func TestVFSProcessSHMRemoval(t *testing.T) {
+ data, meta := processDirs(t)
+ p := launchVFSProcess(t, data, meta)
+ p.ok("open", "")
+ p.ok("exec", "PRAGMA journal_mode=WAL; CREATE TABLE data(value INTEGER); INSERT INTO data VALUES(1)")
+ shm := filepath.Join(data, "main.db-shm")
+ if st, err := os.Stat(shm); err != nil || !st.Mode().IsRegular() {
+ t.Fatalf("SHM was not created: %v", err)
+ }
+ p.ok("close", "")
+ if _, err := os.Stat(shm); !os.IsNotExist(err) {
+ t.Fatalf("custom last close retained SHM: %v", err)
+ }
+ p.ok("open", "")
+ p.count(1)
+ p.stop(false)
+}
diff --git a/internal/vfsspike/vfs_linux.go b/internal/vfsspike/vfs_linux.go
index 4a6ffeeda..d1ee8a7e7 100644
--- a/internal/vfsspike/vfs_linux.go
+++ b/internal/vfsspike/vfs_linux.go
@@ -1,8 +1,8 @@
//go:build linux && (amd64 || arm64)
-// Package vfsspike is an isolated single-connection experiment. It must not be
+// Package vfsspike is an isolated descriptor VFS experiment. It must not be
// mixed with another SQLite VFS on the same inode in this process: POSIX locks
-// are process-owned. Multi-connection bookkeeping is a later spike stage.
+// are process-owned.
package vfsspike
import (
@@ -32,10 +32,15 @@ type VFS struct {
expected identity
ptr, cname, methods uintptr
tls *libc.TLS
- files map[string]*file
+ files map[uintptr]*file
closed bool
rejected []int
beforeOpen func(string)
+ lockHook func(int, int16, int64, int64) error
+ poisoned atomic.Bool
+ retired []*file
+ db dbOwnership
+ shm *sharedMemory
}
type file struct {
v *VFS
@@ -92,7 +97,7 @@ func Register(data, meta *os.File, name string) (*VFS, error) {
}
registry.inodes[id] = true
registry.Unlock()
- v := &VFS{data: -1, meta: -1, main: name, expected: id, files: make(map[string]*file), tls: libc.NewTLS()}
+ v := &VFS{data: -1, meta: -1, main: name, expected: id, files: make(map[uintptr]*file), tls: libc.NewTLS()}
var err error
defer func() {
if err != nil {
@@ -151,6 +156,14 @@ func (v *VFS) Close() error {
if len(v.files) != 0 {
return unix.EBUSY
}
+ if v.poisoned.Load() {
+ if rc := v.finishPoisoned(); rc != sqlite.SQLITE_OK {
+ return unix.EIO
+ }
+ }
+ if v.shm != nil || v.db.readers != 0 || len(v.db.deferred) != 0 {
+ return unix.EBUSY
+ }
if rc := sqlite.Xsqlite3_vfs_unregister(v.tls, v.ptr); rc != sqlite.SQLITE_OK {
return fmt.Errorf("unregister: sqlite code %d", rc)
}
@@ -201,7 +214,7 @@ func openFile(tls *libc.TLS, p, z, out uintptr, flags int32, outFlags uintptr) i
}
v.mu.Lock()
defer v.mu.Unlock()
- if v.closed || v.recheck() != nil {
+ if v.closed || v.poisoned.Load() || v.recheck() != nil {
return sqlite.SQLITE_CANTOPEN
}
name := libc.GoString(z)
@@ -217,9 +230,6 @@ func openFile(tls *libc.TLS, p, z, out uintptr, flags int32, outFlags uintptr) i
} else if !v.allowed(name) {
return sqlite.SQLITE_CANTOPEN
}
- if v.files[name] != nil {
- return sqlite.SQLITE_BUSY
- }
isMain := flags&sqlite.SQLITE_OPEN_MAIN_DB != 0
if isMain && name != v.main {
return sqlite.SQLITE_CANTOPEN
@@ -247,7 +257,7 @@ func openFile(tls *libc.TLS, p, z, out uintptr, flags int32, outFlags uintptr) i
return sqlite.SQLITE_CANTOPEN
}
var s unix.Stat_t
- if err = unix.Fstat(fd, &s); err != nil || s.Mode&unix.S_IFMT != unix.S_IFREG || s.Nlink != 1 || v.alias(&s) || (isMain && !same(&s, v.expected)) {
+ if err = unix.Fstat(fd, &s); err != nil || s.Mode&unix.S_IFMT != unix.S_IFREG || s.Nlink != 1 || v.aliasName(&s, name) || (isMain && !same(&s, v.expected)) {
v.reject(fd)
return sqlite.SQLITE_CANTOPEN
}
@@ -258,7 +268,7 @@ func openFile(tls *libc.TLS, p, z, out uintptr, flags int32, outFlags uintptr) i
return sqlite.SQLITE_CANTOPEN
}
}
- v.files[name] = f
+ v.files[out] = f
registry.Lock()
registry.files[out] = f
registry.Unlock()
@@ -277,23 +287,44 @@ func closeFile(tls *libc.TLS, p uintptr) int32 {
v.mu.Lock()
defer v.mu.Unlock()
rc := int32(sqlite.SQLITE_OK)
- if f.shm >= 0 {
+ if !v.poisoned.Load() && f.shm >= 0 {
rc = unmapLocked(f, false)
}
- err := unix.Close(f.fd)
- delete(v.files, f.name)
- v.drainRejected()
+ if rc != sqlite.SQLITE_OK {
+ v.poisoned.Store(true)
+ }
+ if !v.poisoned.Load() && f.main {
+ rc = unlockDB(f, sqlite.SQLITE_LOCK_NONE)
+ }
+ if rc != sqlite.SQLITE_OK {
+ v.poisoned.Store(true)
+ }
+ if v.poisoned.Load() {
+ v.retired = append(v.retired, f)
+ } else if f.main && v.db.readers > 0 {
+ v.db.deferred = append(v.db.deferred, f.fd)
+ } else if unix.Close(f.fd) != nil {
+ rc = sqlite.SQLITE_IOERR_CLOSE
+ }
+ delete(v.files, p)
registry.Lock()
delete(registry.files, p)
registry.Unlock()
foreignStore(tls, p, sqlite.Tsqlite3_file{})
- if err != nil {
- return sqlite.SQLITE_IOERR_CLOSE
+ if v.poisoned.Load() {
+ if result := v.finishPoisoned(); result != sqlite.SQLITE_OK {
+ rc = result
+ }
+ } else {
+ v.drainRejected()
}
return rc
}
func readFile(tls *libc.TLS, p, b uintptr, n int32, off int64) int32 {
f := getFile(p)
+ if f.v.poisoned.Load() {
+ return sqlite.SQLITE_IOERR_READ
+ }
buf := make([]byte, int(n))
got, err := unix.Pread(f.fd, buf, off)
if err != nil && err != io.EOF {
@@ -308,6 +339,9 @@ func readFile(tls *libc.TLS, p, b uintptr, n int32, off int64) int32 {
}
func writeFile(tls *libc.TLS, p, b uintptr, n int32, off int64) int32 {
f := getFile(p)
+ if f.v.poisoned.Load() {
+ return sqlite.SQLITE_IOERR_WRITE
+ }
buf := make([]byte, int(n))
copyFromForeign(tls, buf, b)
for len(buf) > 0 {
@@ -321,13 +355,20 @@ func writeFile(tls *libc.TLS, p, b uintptr, n int32, off int64) int32 {
return sqlite.SQLITE_OK
}
func truncateFile(tls *libc.TLS, p uintptr, n int64) int32 {
- if unix.Ftruncate(getFile(p).fd, n) != nil {
+ f := getFile(p)
+ if f.v.poisoned.Load() {
+ return sqlite.SQLITE_IOERR_TRUNCATE
+ }
+ if unix.Ftruncate(f.fd, n) != nil {
return sqlite.SQLITE_IOERR_TRUNCATE
}
return sqlite.SQLITE_OK
}
func syncFile(tls *libc.TLS, p uintptr, flags int32) int32 {
f := getFile(p)
+ if f.v.poisoned.Load() {
+ return sqlite.SQLITE_IOERR_FSYNC
+ }
if unix.Fsync(f.fd) != nil {
return sqlite.SQLITE_IOERR_FSYNC
}
@@ -338,7 +379,11 @@ func syncFile(tls *libc.TLS, p uintptr, flags int32) int32 {
}
func sizeFile(tls *libc.TLS, p, out uintptr) int32 {
var s unix.Stat_t
- if unix.Fstat(getFile(p).fd, &s) != nil {
+ f := getFile(p)
+ if f.v.poisoned.Load() {
+ return sqlite.SQLITE_IOERR_FSTAT
+ }
+ if unix.Fstat(f.fd, &s) != nil {
return sqlite.SQLITE_IOERR_FSTAT
}
foreignStore(tls, out, s.Size)
@@ -360,6 +405,9 @@ func accessFile(tls *libc.TLS, p, z uintptr, flags int32, out uintptr) int32 {
foreignStore(tls, out, int32(0))
v := getVFS(p)
name := libc.GoString(z)
+ if v.poisoned.Load() {
+ return sqlite.SQLITE_IOERR_ACCESS
+ }
if !v.allowed(name) {
return sqlite.SQLITE_CANTOPEN
}
@@ -383,7 +431,7 @@ func accessFile(tls *libc.TLS, p, z uintptr, flags int32, out uintptr) int32 {
func deleteFile(tls *libc.TLS, p, z uintptr, syncDir int32) int32 {
v := getVFS(p)
name := libc.GoString(z)
- if !v.allowed(name) || name == v.main || v.recheck() != nil {
+ if v.poisoned.Load() || !v.allowed(name) || name == v.main || v.recheck() != nil {
return sqlite.SQLITE_IOERR_DELETE
}
if err := unix.Unlinkat(v.data, name, 0); err != nil && !errors.Is(err, unix.ENOENT) {
@@ -426,6 +474,9 @@ func sectorSize(*libc.TLS, uintptr) int32 { return 4096 }
func characteristics(*libc.TLS, uintptr) int32 { return 0 }
func controlFile(tls *libc.TLS, p uintptr, op int32, arg uintptr) int32 {
f := getFile(p)
+ if f.v.poisoned.Load() {
+ return sqlite.SQLITE_IOERR_FSTAT
+ }
switch op {
case sqlite.SQLITE_FCNTL_HAS_MOVED:
var moved int32
@@ -456,16 +507,17 @@ func controlFile(tls *libc.TLS, p uintptr, op int32, arg uintptr) int32 {
}
}
-func (v *VFS) alias(s *unix.Stat_t) bool {
+func (v *VFS) aliasName(s *unix.Stat_t, name string) bool {
id := identity{s.Dev, s.Ino}
for _, f := range v.files {
- if f.id == id || (f.shm >= 0 && f.shmID == id) {
+ if (f.id == id && f.name != name) || (f.name == name && f.id != id) {
return true
}
}
- return false
+ return v.shm != nil && v.shm.id == id
}
-func (v *VFS) reject(fd int) { v.rejected = append(v.rejected, fd); v.drainRejected() }
+func (v *VFS) alias(s *unix.Stat_t) bool { return v.aliasName(s, "") }
+func (v *VFS) reject(fd int) { v.rejected = append(v.rejected, fd); v.drainRejected() }
func (v *VFS) drainRejected() {
if len(v.files) != 0 {
return
diff --git a/internal/vfsspike/vfs_linux_test.go b/internal/vfsspike/vfs_linux_test.go
index efdfa386e..9fa72836e 100644
--- a/internal/vfsspike/vfs_linux_test.go
+++ b/internal/vfsspike/vfs_linux_test.go
@@ -11,6 +11,7 @@ import (
"os"
"os/exec"
"path/filepath"
+ "strconv"
"testing"
"unsafe"
)
@@ -112,16 +113,6 @@ func TestRefusesChangedMainAndUnsafeSidecars(t *testing.T) {
})
}
}
-func TestSecondConnectionRefused(t *testing.T) {
- v, _ := setup(t)
- first := connect(t, v)
- execSQL(t, first, "CREATE TABLE data(value)")
- second := connect(t, v)
- if err := second.Ping(); err == nil {
- t.Fatal("second connection accepted")
- }
- execSQL(t, first, "INSERT INTO data VALUES (1)")
-}
func TestDirectoryRenameRetainsAuthority(t *testing.T) {
v, dir := setup(t)
moved := dir + "-moved"
@@ -149,15 +140,40 @@ func TestLockProbeChild(t *testing.T) {
t.Fatal(err)
}
defer unix.Close(fd)
- l := unix.Flock_t{Type: unix.F_WRLCK, Whence: 0, Start: pending + 1, Len: 1}
+ start := pending + 1
+ length := int64(1)
+ want := int64(unix.F_WRLCK)
+ if value := os.Getenv("VFSSPIKE_PROBE_START"); value != "" {
+ start, err = strconv.ParseInt(value, 10, 64)
+ if err != nil {
+ t.Fatal(err)
+ }
+ }
+ if value := os.Getenv("VFSSPIKE_PROBE_LENGTH"); value != "" {
+ length, err = strconv.ParseInt(value, 10, 64)
+ if err != nil {
+ t.Fatal(err)
+ }
+ }
+ if value := os.Getenv("VFSSPIKE_PROBE_WANT"); value != "" {
+ want, err = strconv.ParseInt(value, 10, 16)
+ if err != nil {
+ t.Fatal(err)
+ }
+ }
+ l := unix.Flock_t{Type: unix.F_WRLCK, Whence: 0, Start: start, Len: length}
if err := unix.FcntlFlock(uintptr(fd), unix.F_GETLK, &l); err != nil {
t.Fatal(err)
}
- if l.Type != unix.F_WRLCK {
- t.Fatal("reserved lock disappeared")
+ if l.Type != int16(want) {
+ t.Fatalf("kernel lock type %d want %d at %d", l.Type, want, start)
}
}
func assertReserved(t *testing.T, path string) {
+ t.Helper()
+ assertKernelLock(t, path, pending+1, 1, unix.F_WRLCK)
+}
+func assertKernelLock(t *testing.T, path string, start, length int64, want int16) {
t.Helper()
exe, err := os.Executable()
if err != nil {
@@ -167,7 +183,7 @@ func assertReserved(t *testing.T, path string) {
if runner := os.Getenv("VFSSPIKE_TEST_RUNNER"); runner != "" {
cmd = exec.Command(runner, exe, "-test.run=^TestLockProbeChild$")
}
- cmd.Env = append(os.Environ(), "VFSSPIKE_LOCK_PROBE="+path)
+ cmd.Env = append(os.Environ(), "VFSSPIKE_LOCK_PROBE="+path, "VFSSPIKE_PROBE_START="+strconv.FormatInt(start, 10), "VFSSPIKE_PROBE_LENGTH="+strconv.FormatInt(length, 10), "VFSSPIKE_PROBE_WANT="+strconv.Itoa(int(want)))
if output, err := cmd.CombinedOutput(); err != nil {
t.Fatalf("probe: %v %s", err, output)
}
@@ -183,6 +199,9 @@ func TestRejectedAliasPreservesLocks(t *testing.T) {
db := connect(t, v)
execSQL(t, db, "CREATE TABLE data(value)")
execSQL(t, db, "BEGIN IMMEDIATE")
+ reader := connect(t, v)
+ execSQL(t, reader, "BEGIN")
+ scalar(t, reader, "SELECT count(*) FROM data")
path := filepath.Join(dir, "main.db")
assertReserved(t, path)
link := func() {
@@ -222,6 +241,14 @@ func TestRejectedAliasPreservesLocks(t *testing.T) {
if err := db.Close(); err != nil {
t.Fatal(err)
}
+ assertKernelLock(t, path, pending+2, 510, unix.F_RDLCK)
+ if len(v.rejected) != 1 {
+ t.Fatal("alias descriptor drained under surviving connection")
+ }
+ execSQL(t, reader, "ROLLBACK")
+ if err := reader.Close(); err != nil {
+ t.Fatal(err)
+ }
if len(v.rejected) != 0 {
t.Fatal("rejected descriptors retained after final close")
}