s3blob: fix data race

w.pr cannot be used to indicate that the pipe has been closed. Instead, we use an atomic.Bool.
The problem with using w.pr is that a write to it in the goroutine launched by writer.open
can race with a read on w.pr in Close.

Note that it's still possible that w.pr.CloseWithError runs concurrently with w.pr.Close. But this is fine.

blob/s3blob/s3blob.go

@@ -70,6 +70,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync/atomic"
 
 	s3managerv2 "github.com/aws/aws-sdk-go-v2/feature/s3/manager"
 	s3v2 "github.com/aws/aws-sdk-go-v2/service/s3"
@@ -312,8 +313,9 @@ func (r *reader) Attributes() *driver.ReaderAttributes {
 // writer writes an S3 object, it implements io.WriteCloser.
 type writer struct {
 	// Ends of an io.Pipe, created when the first byte is written.
-	pw *io.PipeWriter
-	pr *io.PipeReader
+	pw       *io.PipeWriter
+	pr       *io.PipeReader
+	prClosed atomic.Bool
 
 	// Alternatively, upload is set to true when Upload was
 	// used to upload data.
@@ -380,7 +382,7 @@ func (w *writer) open(r io.Reader, closePipeOnError bool) {
 		if err != nil {
 			if closePipeOnError {
 				w.pr.CloseWithError(err)
-				w.pr = nil
+				w.prClosed.Store(true)
 			}
 			w.err = err
 		}
@@ -392,7 +394,7 @@ func (w *writer) open(r io.Reader, closePipeOnError bool) {
 // will create an empty file at the given key.
 func (w *writer) Close() error {
 	if !w.upload {
-		if w.pr != nil {
+		if !w.prClosed.Load() {
 			defer w.pr.Close()
 		}
 		if w.pw == nil {