satellite/repair: decouple piece decoding from piece uploading

We have a situation in the repairer where slow uploads cause the error
"internal: quiescence" to be returned from _all_ ongoing uploads,
failing a repair entirely. The quiescence machinery is meant to catch
inactive/stalled _downloads_, but our downloading has already finished
by the time this happens.

It appears that what is happening is this: the piece decoder (the part
that recreates the original segment content) is connected directly to
the piece encoder (the part that encodes new pieces). When the repairer
is doing an upload, it is calling into both of those parts by way of
io.Copy. The quiescence machinery is built into the decoder. When
uploads are especially slow, the piece decoder can't read into the
temporary-file piece contents very far because it can't get too far
ahead (limited buffer size, backpressure, yada yada). This causes the
decoder to complain that it hasn't read anything for 5 seconds, the
threshold for the quiescence error.

It looks like a bit of a complicated job to make an option that disables
quiescence checking in uplink. Instead, we will decouple the decoder and
the encoder in the repairer. Writing the whole segment contents to a
local tempfile through the decoder should make sure that the decoder
doesn't ever observe a 5 second stall (barring some crazy disk problem).
Then the encoder and the uploader can continue from the point of the
reconstructed segment, and it can take as long as it needs to take.

Once it is possible to suppress or avoid the quiescence error in
eestream.decodedReader, we can remove this tempfile step.

Change-Id: I71c68b3460fc4129320364a0514f893e1c47876e

Author: thepaul

satellite/repair/repairer/segments.go

@@ -12,6 +12,7 @@ import (
 	"strconv"
 	"time"
 
+	"github.com/calebcase/tmpfile"
 	"github.com/zeebo/errs"
 	"go.uber.org/zap"
 	"golang.org/x/exp/maps"
@@ -568,6 +569,38 @@ func (repairer *SegmentRepairer) Repair(ctx context.Context, queueSegment queue.
 	}
 	defer func() { err = errs.Combine(err, segmentReader.Close()) }()
 
+	// Reconstruct the segment from the pieces. This should ideally happen in
+	// tandem with the new piece uploads (have Repair() read directly from
+	// segmentReader), but this causes a situation where slow uploads cause
+	// reads from the piece files to block due to backpressure, but then the
+	// slow reads are marked as inactive (a "internal: quiescence" error is
+	// returned). This causes all uploads to fail. Instead, for now, we will
+	// write the reconstructed segment to a tempfile and then upload the pieces
+	// from that.
+	//
+	// Once it is possible to suppress or avoid the quiescence error in
+	// eestream.decodedReader, we can remove this tempfile step.
+	if !repairer.ec.inmemoryDownload {
+		tempfile, err := tmpfile.New("", "repaired-segment-*")
+		if err != nil {
+			return true, repairReconstructError.New("could not open tempfile: %v", err)
+		}
+		_, err = io.Copy(tempfile, segmentReader)
+		if err != nil {
+			return true, repairReconstructError.New("could not reconstruct segment: %v", err)
+		}
+		_, err = tempfile.Seek(0, io.SeekStart)
+		if err != nil {
+			return true, repairReconstructError.New("could not seek to beginning of tempfile: %v", err)
+		}
+		err = segmentReader.Close()
+		// assign to tempfile before returning, because we've already defer-closed segmentReader
+		segmentReader = tempfile
+		if err != nil {
+			return true, repairReconstructError.New("could not close segmentReader: %v", err)
+		}
+	}
+
 	// only report audit result when segment can be successfully downloaded
 	cachedNodesReputation := make(map[storj.NodeID]overlay.ReputationStatus, len(cachedNodesInfo))
 	for id, info := range cachedNodesInfo {