S3 Stream Writer: simpler and more robust logic (#136)

* S3 Stream Writer: simpler and more robust logic

* do not consider context cause for now

* slightly better comments

* remove unused import

* fix typo in comment

* explicit field assign for writeWrapper in NewWriter

* do not overwrite prevErr in Write

* check prevErr before trying to write

* simpler

* avoid passing a ctx

* actually, the ctx will come handy

* pass references to goroutine without params

Author: ahouene

backends/s3/stream.go

@@ -3,9 +3,6 @@ package s3
 import (
 	"context"
 	"io"
-
-	"github.com/PowerDNS/simpleblob"
-	"github.com/minio/minio-go/v7"
 )
 
 // NewReader satisfies StreamReader and provides a read streaming interface to
@@ -22,60 +19,42 @@ func (b *Backend) NewReader(ctx context.Context, name string) (io.ReadCloser, er
 // NewWriter satisfies StreamWriter and provides a write streaming interface to
 // a blob located on an S3 server.
 func (b *Backend) NewWriter(ctx context.Context, name string) (io.WriteCloser, error) {
-	if err := ctx.Err(); err != nil {
-		return nil, err
-	}
+	ctx, cancel := context.WithCancel(ctx)
 	name = b.prependGlobalPrefix(name)
 	pr, pw := io.Pipe()
-	w := &writerWrapper{
-		ctx:      ctx,
-		backend:  b,
-		name:     name,
-		pw:       pw,
-		donePipe: make(chan struct{}),
-	}
 	go func() {
-		var err error
-		// The following call will return only on error or
-		// if the writing end of the pipe is closed.
-		// It is okay to write to w.info from this goroutine
-		// because it will only be used after w.donePipe is closed.
-		w.info, err = w.backend.doStoreReader(w.ctx, w.name, pr, -1)
-		_ = pr.CloseWithError(err) // Always returns nil.
-		close(w.donePipe)
+		// This call returns when the pipe is closed, or when an error occurs.
+		info, err := b.doStoreReader(ctx, name, pr, -1)
+		if err == nil {
+			_ = b.setMarker(ctx, name, info.ETag, false)
+		}
+		_ = pr.CloseWithError(err)
+		cancel()
 	}()
-	return w, nil
+	return &writerWrapper{ctx: ctx, pw: pw}, nil
 }
 
-// A writerWrapper implements io.WriteCloser and is returned by (*Backend).NewWriter.
+// A writerWrapper allows storing data on S3 through a io.WriteCloser.
 type writerWrapper struct {
-	backend *Backend
-
-	// We need to keep these around
-	// to write the marker in Close.
-	ctx  context.Context
-	info minio.UploadInfo
-	name string
-
-	// Writes are sent to this pipe
-	// and then written to S3 in a background goroutine.
-	pw       *io.PipeWriter
-	donePipe chan struct{}
+	ctx context.Context
+	pw  *io.PipeWriter
 }
 
+// Write sends p to store as the S3 object associated with w.
+// An error is returned if Write failed previously, an error occurred in S3, or w is already closed.
 func (w *writerWrapper) Write(p []byte) (int, error) {
 	// Not checking the status of ctx explicitly because it will be propagated
 	// from the reader goroutine.
 	return w.pw.Write(p)
 }
 
+// Close ensures that the written data is saved.
+// An error is returned if Write failed previously, an error occurred in S3, or w is already closed.
 func (w *writerWrapper) Close() error {
-	select {
-	case <-w.donePipe:
-		return simpleblob.ErrClosed
-	default:
-	}
-	_ = w.pw.Close() // Always returns nil.
-	<-w.donePipe     // Wait for doStoreReader to return and w.info to be set.
-	return w.backend.setMarker(w.ctx, w.name, w.info.ETag, false)
+	_, err := w.pw.Write(nil)
+	_ = w.pw.Close()
+	// Let the reading goroutine finish writing,
+	// and write the marker if needed.
+	<-w.ctx.Done() // cancelled after writing the marker
+	return err
 }