Nick Craig-Wood
16e199067e
Previously if a cached file had grown larger than the remote object and the cached range metadata was out of sync with the cache file (e.g. after an unclean shutdown) - reloading the file failed with an "invalid seek position" error. This aborted the writeback and left the file inaccessible. Rclone now recovers the bytes that are still available from the remote and logs an ERROR that the local file is likely corrupted after an interrupted upload. See #9231.
826 lines
22 KiB
Go
826 lines
22 KiB
Go
package vfscache
|
|
|
|
// FIXME need to test async writeback here
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"io"
|
|
"math/rand"
|
|
"os"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/rclone/rclone/fs"
|
|
"github.com/rclone/rclone/fstest"
|
|
"github.com/rclone/rclone/lib/random"
|
|
"github.com/rclone/rclone/lib/ranges"
|
|
"github.com/rclone/rclone/lib/readers"
|
|
"github.com/rclone/rclone/vfs/vfscommon"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
)
|
|
|
|
var zeroes = string(make([]byte, 100))
|
|
|
|
func newItemTestCache(t *testing.T) (r *fstest.Run, c *Cache) {
|
|
opt := vfscommon.Opt
|
|
|
|
// Disable the cache cleaner as it interferes with these tests
|
|
opt.CachePollInterval = 0
|
|
|
|
// Disable synchronous write
|
|
opt.WriteBack = 0
|
|
|
|
// Disable handle caching so existing tests get immediate close behavior
|
|
opt.HandleCaching = 0
|
|
|
|
return newTestCacheOpt(t, opt)
|
|
}
|
|
|
|
// Check the object has contents
|
|
func checkObject(t *testing.T, r *fstest.Run, remote string, contents string) {
|
|
obj, err := r.Fremote.NewObject(context.Background(), remote)
|
|
require.NoError(t, err)
|
|
in, err := obj.Open(context.Background())
|
|
require.NoError(t, err)
|
|
buf, err := io.ReadAll(in)
|
|
require.NoError(t, err)
|
|
require.NoError(t, in.Close())
|
|
assert.Equal(t, contents, string(buf))
|
|
}
|
|
|
|
func newFileLength(t *testing.T, r *fstest.Run, c *Cache, remote string, length int) (contents string, obj fs.Object, item *Item) {
|
|
contents = random.String(length)
|
|
r.WriteObject(context.Background(), remote, contents, time.Now())
|
|
item, _ = c.get(remote)
|
|
obj, err := r.Fremote.NewObject(context.Background(), remote)
|
|
require.NoError(t, err)
|
|
return
|
|
}
|
|
|
|
func newFile(t *testing.T, r *fstest.Run, c *Cache, remote string) (contents string, obj fs.Object, item *Item) {
|
|
return newFileLength(t, r, c, remote, 100)
|
|
}
|
|
|
|
func TestItemExists(t *testing.T) {
|
|
_, c := newItemTestCache(t)
|
|
item, _ := c.get("potato")
|
|
|
|
assert.False(t, item.Exists())
|
|
require.NoError(t, item.Open(nil))
|
|
assert.True(t, item.Exists())
|
|
require.NoError(t, item.Close(nil))
|
|
assert.True(t, item.Exists())
|
|
item.remove("test")
|
|
assert.False(t, item.Exists())
|
|
}
|
|
|
|
func TestItemGetSize(t *testing.T) {
|
|
r, c := newItemTestCache(t)
|
|
item, _ := c.get("potato")
|
|
require.NoError(t, item.Open(nil))
|
|
|
|
size, err := item.GetSize()
|
|
require.NoError(t, err)
|
|
assert.Equal(t, int64(0), size)
|
|
|
|
n, err := item.WriteAt([]byte("hello"), 0)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, 5, n)
|
|
|
|
size, err = item.GetSize()
|
|
require.NoError(t, err)
|
|
assert.Equal(t, int64(5), size)
|
|
|
|
require.NoError(t, item.Close(nil))
|
|
checkObject(t, r, "potato", "hello")
|
|
}
|
|
|
|
func TestItemDirty(t *testing.T) {
|
|
r, c := newItemTestCache(t)
|
|
item, _ := c.get("potato")
|
|
require.NoError(t, item.Open(nil))
|
|
|
|
assert.Equal(t, false, item.IsDirty())
|
|
|
|
n, err := item.WriteAt([]byte("hello"), 0)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, 5, n)
|
|
|
|
assert.Equal(t, true, item.IsDirty())
|
|
|
|
require.NoError(t, item.Close(nil))
|
|
|
|
// Sync writeback so expect clean here
|
|
assert.Equal(t, false, item.IsDirty())
|
|
|
|
item.Dirty()
|
|
|
|
assert.Equal(t, true, item.IsDirty())
|
|
checkObject(t, r, "potato", "hello")
|
|
}
|
|
|
|
func TestItemSync(t *testing.T) {
|
|
_, c := newItemTestCache(t)
|
|
item, _ := c.get("potato")
|
|
|
|
require.Error(t, item.Sync())
|
|
|
|
require.NoError(t, item.Open(nil))
|
|
|
|
require.NoError(t, item.Sync())
|
|
|
|
require.NoError(t, item.Close(nil))
|
|
}
|
|
|
|
func TestItemTruncateNew(t *testing.T) {
|
|
r, c := newItemTestCache(t)
|
|
item, _ := c.get("potato")
|
|
|
|
require.Error(t, item.Truncate(0))
|
|
|
|
require.NoError(t, item.Open(nil))
|
|
|
|
require.NoError(t, item.Truncate(100))
|
|
|
|
size, err := item.GetSize()
|
|
require.NoError(t, err)
|
|
assert.Equal(t, int64(100), size)
|
|
|
|
// Check the Close callback works
|
|
callbackCalled := false
|
|
callback := func(o fs.Object) {
|
|
callbackCalled = true
|
|
assert.Equal(t, "potato", o.Remote())
|
|
assert.Equal(t, int64(100), o.Size())
|
|
}
|
|
require.NoError(t, item.Close(callback))
|
|
assert.True(t, callbackCalled)
|
|
|
|
checkObject(t, r, "potato", zeroes)
|
|
}
|
|
|
|
func TestItemTruncateExisting(t *testing.T) {
|
|
r, c := newItemTestCache(t)
|
|
|
|
contents, obj, item := newFile(t, r, c, "existing")
|
|
|
|
require.Error(t, item.Truncate(40))
|
|
checkObject(t, r, "existing", contents)
|
|
|
|
require.NoError(t, item.Open(obj))
|
|
|
|
require.NoError(t, item.Truncate(40))
|
|
|
|
require.NoError(t, item.Truncate(60))
|
|
|
|
require.NoError(t, item.Close(nil))
|
|
|
|
checkObject(t, r, "existing", contents[:40]+zeroes[:20])
|
|
}
|
|
|
|
func TestItemReadAt(t *testing.T) {
|
|
r, c := newItemTestCache(t)
|
|
|
|
contents, obj, item := newFile(t, r, c, "existing")
|
|
buf := make([]byte, 10)
|
|
|
|
_, err := item.ReadAt(buf, 10)
|
|
require.Error(t, err)
|
|
|
|
require.NoError(t, item.Open(obj))
|
|
|
|
n, err := item.ReadAt(buf, 10)
|
|
assert.Equal(t, 10, n)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, contents[10:20], string(buf[:n]))
|
|
|
|
n, err = item.ReadAt(buf, 95)
|
|
assert.Equal(t, 5, n)
|
|
assert.Equal(t, io.EOF, err)
|
|
assert.Equal(t, contents[95:], string(buf[:n]))
|
|
|
|
n, err = item.ReadAt(buf, 1000)
|
|
assert.Equal(t, 0, n)
|
|
assert.Equal(t, io.EOF, err)
|
|
assert.Equal(t, contents[:0], string(buf[:n]))
|
|
|
|
n, err = item.ReadAt(buf, -1)
|
|
assert.Equal(t, 0, n)
|
|
assert.Equal(t, io.EOF, err)
|
|
assert.Equal(t, contents[:0], string(buf[:n]))
|
|
|
|
require.NoError(t, item.Close(nil))
|
|
}
|
|
|
|
func TestItemWriteAtNew(t *testing.T) {
|
|
r, c := newItemTestCache(t)
|
|
item, _ := c.get("potato")
|
|
buf := make([]byte, 10)
|
|
|
|
_, err := item.WriteAt(buf, 10)
|
|
require.Error(t, err)
|
|
|
|
require.NoError(t, item.Open(nil))
|
|
|
|
assert.Equal(t, int64(0), item.getDiskSize())
|
|
|
|
n, err := item.WriteAt([]byte("HELLO"), 10)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, 5, n)
|
|
|
|
// FIXME we account for the sparse data we've "written" to
|
|
// disk here so this is actually 5 bytes higher than expected
|
|
assert.Equal(t, int64(15), item.getDiskSize())
|
|
|
|
n, err = item.WriteAt([]byte("THEND"), 20)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, 5, n)
|
|
|
|
assert.Equal(t, int64(25), item.getDiskSize())
|
|
|
|
require.NoError(t, item.Close(nil))
|
|
|
|
checkObject(t, r, "potato", zeroes[:10]+"HELLO"+zeroes[:5]+"THEND")
|
|
}
|
|
|
|
func TestItemWriteAtExisting(t *testing.T) {
|
|
r, c := newItemTestCache(t)
|
|
|
|
contents, obj, item := newFile(t, r, c, "existing")
|
|
|
|
require.NoError(t, item.Open(obj))
|
|
|
|
n, err := item.WriteAt([]byte("HELLO"), 10)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, 5, n)
|
|
|
|
n, err = item.WriteAt([]byte("THEND"), 95)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, 5, n)
|
|
|
|
n, err = item.WriteAt([]byte("THEVERYEND"), 120)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, 10, n)
|
|
|
|
require.NoError(t, item.Close(nil))
|
|
|
|
checkObject(t, r, "existing", contents[:10]+"HELLO"+contents[15:95]+"THEND"+zeroes[:20]+"THEVERYEND")
|
|
}
|
|
|
|
func TestItemLoadMeta(t *testing.T) {
|
|
r, c := newItemTestCache(t)
|
|
|
|
contents, obj, item := newFile(t, r, c, "existing")
|
|
_ = contents
|
|
|
|
// Open the object to create metadata for it
|
|
require.NoError(t, item.Open(obj))
|
|
require.NoError(t, item.Close(nil))
|
|
info := item.info
|
|
|
|
// Remove the item from the cache
|
|
c.mu.Lock()
|
|
delete(c.item, item.name)
|
|
c.mu.Unlock()
|
|
|
|
// Reload the item so we have to load the metadata
|
|
item2, _ := c._get("existing")
|
|
require.NoError(t, item2.Open(obj))
|
|
info2 := item.info
|
|
require.NoError(t, item2.Close(nil))
|
|
|
|
// Check that the item is different
|
|
assert.NotEqual(t, item, item2)
|
|
// ... but the info is the same
|
|
assert.Equal(t, info, info2)
|
|
}
|
|
|
|
func TestItemReload(t *testing.T) {
|
|
r, c := newItemTestCache(t)
|
|
|
|
contents, obj, item := newFile(t, r, c, "existing")
|
|
_ = contents
|
|
|
|
// Open the object to create metadata for it
|
|
require.NoError(t, item.Open(obj))
|
|
|
|
// Make it dirty
|
|
n, err := item.WriteAt([]byte("THEENDMYFRIEND"), 95)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, 14, n)
|
|
assert.True(t, item.IsDirty())
|
|
|
|
// Close the file to pacify Windows, but don't call item.Close()
|
|
item.mu.Lock()
|
|
require.NoError(t, item.fd.Close())
|
|
item.fd = nil
|
|
item.mu.Unlock()
|
|
|
|
// Remove the item from the cache
|
|
c.mu.Lock()
|
|
delete(c.item, item.name)
|
|
c.mu.Unlock()
|
|
|
|
// Reload the item so we have to load the metadata and restart
|
|
// the transfer
|
|
item2, _ := c._get("existing")
|
|
require.NoError(t, item2.reload(context.Background()))
|
|
assert.False(t, item2.IsDirty())
|
|
|
|
// Check that the item is different
|
|
assert.NotEqual(t, item, item2)
|
|
|
|
// And check the contents got written back to the remote
|
|
checkObject(t, r, "existing", contents[:95]+"THEENDMYFRIEND")
|
|
|
|
// And check that AddVirtual was called
|
|
assert.Equal(t, []avInfo{
|
|
{Remote: "existing", Size: 109, IsDir: false},
|
|
}, avInfos)
|
|
}
|
|
|
|
func TestItemReloadRemoteGone(t *testing.T) {
|
|
r, c := newItemTestCache(t)
|
|
|
|
contents, obj, item := newFile(t, r, c, "existing")
|
|
_ = contents
|
|
|
|
// Open the object to create metadata for it
|
|
require.NoError(t, item.Open(obj))
|
|
|
|
size, err := item.GetSize()
|
|
require.NoError(t, err)
|
|
assert.Equal(t, int64(100), size)
|
|
|
|
// Read something to instantiate the cache file
|
|
buf := make([]byte, 10)
|
|
_, err = item.ReadAt(buf, 10)
|
|
require.NoError(t, err)
|
|
|
|
// Test cache file present
|
|
_, err = os.Stat(item.c.toOSPath(item.name))
|
|
require.NoError(t, err)
|
|
|
|
require.NoError(t, item.Close(nil))
|
|
|
|
// Remove the remote object
|
|
require.NoError(t, obj.Remove(context.Background()))
|
|
|
|
// Re-open with no object
|
|
require.NoError(t, item.Open(nil))
|
|
|
|
// Check size is now 0
|
|
size, err = item.GetSize()
|
|
require.NoError(t, err)
|
|
assert.Equal(t, int64(0), size)
|
|
|
|
// Test cache file is now empty
|
|
fi, err := os.Stat(item.c.toOSPath(item.name))
|
|
require.NoError(t, err)
|
|
assert.Equal(t, int64(0), fi.Size())
|
|
|
|
require.NoError(t, item.Close(nil))
|
|
}
|
|
|
|
func TestItemReloadCacheStale(t *testing.T) {
|
|
r, c := newItemTestCache(t)
|
|
|
|
contents, obj, item := newFile(t, r, c, "existing")
|
|
|
|
// Open the object to create metadata for it
|
|
require.NoError(t, item.Open(obj))
|
|
|
|
size, err := item.GetSize()
|
|
require.NoError(t, err)
|
|
assert.Equal(t, int64(100), size)
|
|
|
|
// Read something to instantiate the cache file
|
|
buf := make([]byte, 10)
|
|
_, err = item.ReadAt(buf, 10)
|
|
require.NoError(t, err)
|
|
|
|
// Test cache file present
|
|
_, err = os.Stat(item.c.toOSPath(item.name))
|
|
require.NoError(t, err)
|
|
|
|
require.NoError(t, item.Close(nil))
|
|
|
|
// Update the remote to something different
|
|
contents2, obj, item := newFileLength(t, r, c, "existing", 110)
|
|
assert.NotEqual(t, contents, contents2)
|
|
|
|
// Re-open with updated object
|
|
oldFingerprint := item.info.Fingerprint
|
|
assert.NotEqual(t, "", oldFingerprint)
|
|
require.NoError(t, item.Open(obj))
|
|
|
|
// Make sure fingerprint was updated
|
|
assert.NotEqual(t, oldFingerprint, item.info.Fingerprint)
|
|
assert.NotEqual(t, "", item.info.Fingerprint)
|
|
|
|
// Check size is now 110
|
|
size, err = item.GetSize()
|
|
require.NoError(t, err)
|
|
assert.Equal(t, int64(110), size)
|
|
|
|
// Test cache file is now correct size
|
|
fi, err := os.Stat(item.c.toOSPath(item.name))
|
|
require.NoError(t, err)
|
|
assert.Equal(t, int64(110), fi.Size())
|
|
|
|
// Write to the file to make it dirty
|
|
// This checks we aren't reusing stale data
|
|
n, err := item.WriteAt([]byte("HELLO"), 0)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, 5, n)
|
|
assert.Equal(t, true, item.IsDirty())
|
|
|
|
require.NoError(t, item.Close(nil))
|
|
|
|
// Now check with all that swizzling stuff around that the
|
|
// object is correct
|
|
|
|
checkObject(t, r, "existing", "HELLO"+contents2[5:])
|
|
}
|
|
|
|
// TestItemReloadDirtyBeyondRemote checks that reloading a dirty cache
|
|
// item whose cache file has grown larger than the remote object
|
|
// recovers what it can from the cache file instead of failing with
|
|
// "invalid seek position".
|
|
func TestItemReloadDirtyBeyondRemote(t *testing.T) {
|
|
r, c := newItemTestCache(t)
|
|
|
|
// Small remote object (100 bytes)
|
|
_, obj, item := newFile(t, r, c, "existing")
|
|
|
|
// Open it and write over and past the end of the remote object,
|
|
// growing the cache file to 200 bytes and making it dirty.
|
|
require.NoError(t, item.Open(obj))
|
|
newContents := random.String(200)
|
|
n, err := item.WriteAt([]byte(newContents), 0)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, 200, n)
|
|
assert.True(t, item.IsDirty())
|
|
|
|
size, err := item.GetSize()
|
|
require.NoError(t, err)
|
|
assert.Equal(t, int64(200), size)
|
|
|
|
// Simulate an unclean shutdown: the cache file on disk is fully grown
|
|
// but the final range-metadata update was never persisted, so Rs
|
|
// under-reports the parts of the file that are present. Here the
|
|
// remaining present range [0,150) ends beyond the 100 byte remote
|
|
// object, so the missing range [150,200) starts past the end of the
|
|
// remote object.
|
|
item.mu.Lock()
|
|
item.info.Rs = item.info.Rs.Intersection(ranges.Range{Pos: 0, Size: 150})
|
|
require.NoError(t, item._save())
|
|
require.NoError(t, item.fd.Close())
|
|
item.fd = nil
|
|
item.mu.Unlock()
|
|
|
|
// Drop the item so reload has to reload the metadata from disk
|
|
c.mu.Lock()
|
|
delete(c.item, item.name)
|
|
c.mu.Unlock()
|
|
|
|
// Reload the dirty item. Before the fix this failed trying to download
|
|
// the missing [150,200) range from the 100 byte remote object with
|
|
// "invalid seek position".
|
|
item2, _ := c._get("existing")
|
|
require.NoError(t, item2.reload(context.Background()))
|
|
assert.False(t, item2.IsDirty())
|
|
|
|
// The grown contents are recovered from the cache file and written
|
|
// back to the remote.
|
|
checkObject(t, r, "existing", newContents)
|
|
}
|
|
|
|
func TestItemReadWrite(t *testing.T) {
|
|
r, c := newItemTestCache(t)
|
|
const (
|
|
size = 50*1024*1024 + 123
|
|
fileName = "large"
|
|
)
|
|
|
|
item, _ := c.get(fileName)
|
|
require.NoError(t, item.Open(nil))
|
|
|
|
// Create the test file
|
|
in := readers.NewPatternReader(size)
|
|
buf := make([]byte, 1024*1024)
|
|
buf2 := make([]byte, 1024*1024)
|
|
offset := int64(0)
|
|
for {
|
|
n, err := in.Read(buf)
|
|
n2, err2 := item.WriteAt(buf[:n], offset)
|
|
offset += int64(n2)
|
|
require.NoError(t, err2)
|
|
require.Equal(t, n, n2)
|
|
if err == io.EOF {
|
|
break
|
|
}
|
|
require.NoError(t, err)
|
|
}
|
|
|
|
// Check it is the right size
|
|
readSize, err := item.GetSize()
|
|
require.NoError(t, err)
|
|
assert.Equal(t, int64(size), readSize)
|
|
|
|
require.NoError(t, item.Close(nil))
|
|
|
|
assert.False(t, item.remove(fileName))
|
|
|
|
obj, err := r.Fremote.NewObject(context.Background(), fileName)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, int64(size), obj.Size())
|
|
|
|
// read and check a block of size N at offset
|
|
// It returns eof true if the end of file has been reached
|
|
readCheckBuf := func(t *testing.T, in io.ReadSeeker, buf, buf2 []byte, item *Item, offset int64, N int) (n int, eof bool) {
|
|
what := fmt.Sprintf("buf=%p, buf2=%p, item=%p, offset=%d, N=%d", buf, buf2, item, offset, N)
|
|
n, err := item.ReadAt(buf, offset)
|
|
|
|
_, err2 := in.Seek(offset, io.SeekStart)
|
|
require.NoError(t, err2, what)
|
|
n2, err2 := in.Read(buf2[:n])
|
|
require.Equal(t, n, n2, what)
|
|
assert.Equal(t, buf[:n], buf2[:n2], what)
|
|
assert.Equal(t, buf[:n], buf2[:n2], what)
|
|
|
|
if err == io.EOF {
|
|
return n, true
|
|
}
|
|
require.NoError(t, err, what)
|
|
require.NoError(t, err2, what)
|
|
return n, false
|
|
}
|
|
readCheck := func(t *testing.T, item *Item, offset int64, N int) (n int, eof bool) {
|
|
return readCheckBuf(t, in, buf, buf2, item, offset, N)
|
|
}
|
|
|
|
// Read it back sequentially
|
|
t.Run("Sequential", func(t *testing.T) {
|
|
require.NoError(t, item.Open(obj))
|
|
assert.False(t, item.present())
|
|
offset := int64(0)
|
|
for {
|
|
n, eof := readCheck(t, item, offset, len(buf))
|
|
offset += int64(n)
|
|
if eof {
|
|
break
|
|
}
|
|
}
|
|
assert.Equal(t, int64(size), offset)
|
|
require.NoError(t, item.Close(nil))
|
|
assert.False(t, item.remove(fileName))
|
|
})
|
|
|
|
// Read it back randomly
|
|
t.Run("Random", func(t *testing.T) {
|
|
require.NoError(t, item.Open(obj))
|
|
assert.False(t, item.present())
|
|
for !item.present() {
|
|
blockSize := rand.Intn(len(buf))
|
|
offset := max(rand.Int63n(size+2*int64(blockSize))-int64(blockSize), 0)
|
|
_, _ = readCheck(t, item, offset, blockSize)
|
|
}
|
|
require.NoError(t, item.Close(nil))
|
|
assert.False(t, item.remove(fileName))
|
|
})
|
|
|
|
// Read it back randomly concurrently
|
|
t.Run("RandomConcurrent", func(t *testing.T) {
|
|
require.NoError(t, item.Open(obj))
|
|
assert.False(t, item.present())
|
|
var wg sync.WaitGroup
|
|
for range 8 {
|
|
wg.Go(func() {
|
|
in := readers.NewPatternReader(size)
|
|
buf := make([]byte, 1024*1024)
|
|
buf2 := make([]byte, 1024*1024)
|
|
for !item.present() {
|
|
blockSize := rand.Intn(len(buf))
|
|
offset := max(rand.Int63n(size+2*int64(blockSize))-int64(blockSize), 0)
|
|
_, _ = readCheckBuf(t, in, buf, buf2, item, offset, blockSize)
|
|
}
|
|
})
|
|
}
|
|
wg.Wait()
|
|
require.NoError(t, item.Close(nil))
|
|
assert.False(t, item.remove(fileName))
|
|
})
|
|
|
|
// Read it back in reverse which creates the maximum number of
|
|
// downloaders
|
|
t.Run("Reverse", func(t *testing.T) {
|
|
require.NoError(t, item.Open(obj))
|
|
assert.False(t, item.present())
|
|
offset := int64(size)
|
|
for {
|
|
blockSize := len(buf)
|
|
offset -= int64(blockSize)
|
|
if offset < 0 {
|
|
offset = 0
|
|
blockSize += int(offset)
|
|
}
|
|
_, _ = readCheck(t, item, offset, blockSize)
|
|
if offset == 0 {
|
|
break
|
|
}
|
|
}
|
|
require.NoError(t, item.Close(nil))
|
|
assert.False(t, item.remove(fileName))
|
|
})
|
|
}
|
|
|
|
func newItemTestCacheHandleCaching(t *testing.T, handleCaching time.Duration) (r *fstest.Run, c *Cache) {
|
|
opt := vfscommon.Opt
|
|
|
|
// Disable the cache cleaner as it interferes with these tests
|
|
opt.CachePollInterval = 0
|
|
|
|
// Disable synchronous write
|
|
opt.WriteBack = 0
|
|
|
|
// Set handle caching grace period
|
|
opt.HandleCaching = fs.Duration(handleCaching)
|
|
|
|
return newTestCacheOpt(t, opt)
|
|
}
|
|
|
|
func TestItemHandleCaching(t *testing.T) {
|
|
r, c := newItemTestCacheHandleCaching(t, 1*time.Second)
|
|
|
|
contents, obj, item := newFile(t, r, c, "existing")
|
|
|
|
// Open, read, and close the item
|
|
require.NoError(t, item.Open(obj))
|
|
|
|
buf := make([]byte, 10)
|
|
n, err := item.ReadAt(buf, 0)
|
|
assert.Equal(t, 10, n)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, contents[:10], string(buf[:n]))
|
|
|
|
require.NoError(t, item.Close(nil))
|
|
|
|
// After close, grace period should keep fd and downloaders alive
|
|
item.mu.Lock()
|
|
assert.NotNil(t, item.fd, "fd should still be open during grace period")
|
|
assert.NotNil(t, item.downloaders, "downloaders should still be alive during grace period")
|
|
assert.NotNil(t, item.graceTimer, "grace timer should be set")
|
|
item.mu.Unlock()
|
|
|
|
// Re-open the item - should reuse existing fd and downloaders
|
|
require.NoError(t, item.Open(obj))
|
|
|
|
// Read data to verify it works
|
|
n, err = item.ReadAt(buf, 10)
|
|
assert.Equal(t, 10, n)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, contents[10:20], string(buf[:n]))
|
|
|
|
// Close again
|
|
require.NoError(t, item.Close(nil))
|
|
|
|
// Wait for grace period to expire
|
|
time.Sleep(1500 * time.Millisecond)
|
|
|
|
// After grace period, fd and downloaders should be cleaned up
|
|
item.mu.Lock()
|
|
assert.Nil(t, item.fd, "fd should be closed after grace period")
|
|
assert.Nil(t, item.downloaders, "downloaders should be closed after grace period")
|
|
assert.Nil(t, item.graceTimer, "grace timer should be nil after expiry")
|
|
item.mu.Unlock()
|
|
}
|
|
|
|
func TestItemHandleCachingDisabled(t *testing.T) {
|
|
r, c := newItemTestCacheHandleCaching(t, 0)
|
|
|
|
contents, obj, item := newFile(t, r, c, "existing")
|
|
_ = contents
|
|
|
|
// Open and close the item
|
|
require.NoError(t, item.Open(obj))
|
|
require.NoError(t, item.Close(nil))
|
|
|
|
// With handle caching disabled, fd and downloaders should be immediately closed
|
|
item.mu.Lock()
|
|
assert.Nil(t, item.fd, "fd should be closed immediately when handle caching disabled")
|
|
assert.Nil(t, item.downloaders, "downloaders should be closed immediately when handle caching disabled")
|
|
assert.Nil(t, item.graceTimer, "grace timer should not be set when handle caching disabled")
|
|
item.mu.Unlock()
|
|
}
|
|
|
|
func TestItemHandleCachingReset(t *testing.T) {
|
|
r, c := newItemTestCacheHandleCaching(t, 1*time.Second)
|
|
|
|
_, obj, item := newFile(t, r, c, "existing")
|
|
|
|
// Open, read (to instantiate cache), and close the item
|
|
require.NoError(t, item.Open(obj))
|
|
|
|
buf := make([]byte, 10)
|
|
_, err := item.ReadAt(buf, 0)
|
|
require.NoError(t, err)
|
|
|
|
require.NoError(t, item.Close(nil))
|
|
|
|
// Grace timer should be active
|
|
item.mu.Lock()
|
|
assert.NotNil(t, item.graceTimer, "grace timer should be set")
|
|
item.mu.Unlock()
|
|
|
|
// Reset should skip the item during grace period
|
|
rr, _, err := item.Reset()
|
|
require.NoError(t, err)
|
|
assert.Equal(t, SkippedGrace, rr)
|
|
|
|
// Grace timer should still be active
|
|
item.mu.Lock()
|
|
assert.NotNil(t, item.graceTimer, "grace timer should still be set after skipped reset")
|
|
item.mu.Unlock()
|
|
|
|
// Wait for grace period to expire then reset should remove the item
|
|
time.Sleep(1500 * time.Millisecond)
|
|
|
|
rr, _, err = item.Reset()
|
|
require.NoError(t, err)
|
|
assert.Equal(t, RemovedNotInUse, rr)
|
|
}
|
|
|
|
// TestItemHandleCachingReopenDuringGraceClose reproduces a race between
|
|
// reopening an item and the grace-period close firing for it.
|
|
//
|
|
// closeAfterGrace clears the grace timer and then runs the actual close,
|
|
// which temporarily drops item.mu while it tears down the downloaders -
|
|
// at that point the file handle is still open. A reopen landing in that
|
|
// window used to see no grace timer and a live fd and fail _createFile
|
|
// with "internal error: didn't Close file".
|
|
func TestItemHandleCachingReopenDuringGraceClose(t *testing.T) {
|
|
r, c := newItemTestCacheHandleCaching(t, 10*time.Second)
|
|
|
|
_, obj, item := newFile(t, r, c, "existing")
|
|
buf := make([]byte, 1)
|
|
|
|
const iterations = 50
|
|
for i := 0; i < iterations; i++ {
|
|
// Open, read (to create a downloader) and close so a grace
|
|
// timer is pending with the fd and downloaders still alive.
|
|
require.NoError(t, item.Open(obj))
|
|
_, err := item.ReadAt(buf, 0)
|
|
require.NoError(t, err)
|
|
require.NoError(t, item.Close(nil))
|
|
|
|
// Drive the grace close ourselves so we can race it against a
|
|
// reopen. Hold item.mu and park both the close (A) and the
|
|
// reopen (B) on the lock with A queued first. When we release,
|
|
// A runs the close, which drops item.mu to tear down the
|
|
// downloaders, and B - already waiting - grabs it in that
|
|
// window and observes the still-open fd.
|
|
item.mu.Lock()
|
|
require.NotNil(t, item.graceTimer, "grace timer should be set after close")
|
|
item.graceTimer.Stop()
|
|
|
|
var openErr error
|
|
var wg sync.WaitGroup
|
|
wg.Add(2)
|
|
startA := make(chan struct{})
|
|
startB := make(chan struct{})
|
|
go func() {
|
|
defer wg.Done()
|
|
<-startA
|
|
item.closeAfterGrace()
|
|
}()
|
|
go func() {
|
|
defer wg.Done()
|
|
<-startB
|
|
openErr = item.Open(obj)
|
|
}()
|
|
close(startA)
|
|
time.Sleep(time.Millisecond) // let A park on item.mu first
|
|
close(startB)
|
|
time.Sleep(time.Millisecond) // let B park on item.mu
|
|
item.mu.Unlock()
|
|
wg.Wait()
|
|
|
|
require.NoError(t, openErr, "reopen racing a grace-period close failed on iteration %d", i)
|
|
|
|
// Drop the handle from the successful reopen so the next
|
|
// iteration starts from a closed item.
|
|
require.NoError(t, item.Close(nil))
|
|
}
|
|
|
|
// Stop the grace timer left pending by the final close.
|
|
item.mu.Lock()
|
|
if item.graceTimer != nil {
|
|
item.graceTimer.Stop()
|
|
}
|
|
item.mu.Unlock()
|
|
}
|