Compare commits

..
4 changed files with 100 additions and 1 deletions
+1 -1
View File
@@ -72,7 +72,7 @@ func (s *localStorage) Get(ctx context.Context, name string) ([]byte, error) {
if err != nil {
return nil, err
}
data, err := os.ReadFile(full)
data, err := readFile(ctx, full)
if err != nil {
if os.IsNotExist(err) {
return nil, errNotFound
+12
View File
@@ -0,0 +1,12 @@
//go:build !windows
package xdrive
import (
"context"
"os"
)
func readFile(_ context.Context, name string) ([]byte, error) {
return os.ReadFile(name)
}
@@ -0,0 +1,30 @@
package xdrive
import (
"context"
"errors"
"os"
"time"
"golang.org/x/sys/windows"
)
// readFile retries os.ReadFile for up to 2 seconds on sharing violations:
// os.Rename (MoveFileEx) in Put holds the file with DELETE access, which
// os.ReadFile does not share, until a moment after its new name appears in List
func readFile(ctx context.Context, name string) ([]byte, error) {
deadline := time.Now().Add(2 * time.Second)
delay := time.Millisecond
for {
data, err := os.ReadFile(name)
if !errors.Is(err, windows.ERROR_SHARING_VIOLATION) || time.Now().Add(delay).After(deadline) {
return data, err
}
select {
case <-ctx.Done():
return nil, err
case <-time.After(delay):
}
delay = min(2*delay, 100*time.Millisecond)
}
}
@@ -0,0 +1,57 @@
package xdrive
import (
"context"
"errors"
"sync"
"testing"
"time"
"golang.org/x/sys/windows"
)
func TestLocalSharingViolation(t *testing.T) {
storage, err := newLocalStorage(t.TempDir())
if err != nil {
t.Fatalf("newLocalStorage: %v", err)
}
const name = "streams/session/c2s/000000000.seg"
if err := storage.Put(context.Background(), name, []byte("segment")); err != nil {
t.Fatalf("Put: %v", err)
}
full, err := storage.resolve(name)
if err != nil {
t.Fatalf("resolve: %v", err)
}
path, err := windows.UTF16PtrFromString(full)
if err != nil {
t.Fatalf("UTF16PtrFromString: %v", err)
}
// os.Rename holds the file like this until a moment after its new name
// appears
handle, err := windows.CreateFile(path, windows.DELETE,
windows.FILE_SHARE_READ|windows.FILE_SHARE_WRITE|windows.FILE_SHARE_DELETE,
nil, windows.OPEN_EXISTING, 0, 0)
if err != nil {
t.Fatalf("CreateFile: %v", err)
}
var once sync.Once
release := func() { once.Do(func() { windows.CloseHandle(handle) }) }
defer release()
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()
if _, err := storage.Get(ctx, name); !errors.Is(err, windows.ERROR_SHARING_VIOLATION) {
t.Fatalf("Get returned %v while the file is held, want a sharing violation", err)
}
time.AfterFunc(100*time.Millisecond, release)
data, err := storage.Get(context.Background(), name)
if err != nil {
t.Fatalf("Get: %v", err)
}
if string(data) != "segment" {
t.Fatalf("Get returned %q, want %q", data, "segment")
}
}