feat(storage): add CurrentState function to determine state of stream in MRD (googleapis#11688)

* feat(storage): add a new function to determine state of mrd

* change method name and output error + add a bit of testing

* change func name

---------

Co-authored-by: Chris Cotter <cjcotter@google.com>
Co-authored-by: BrennaEpp <brennae@google.com>

Author: 3 people

storage/client_test.go

@@ -1459,6 +1459,11 @@ func TestMultiRangeDownloaderEmulated(t *testing.T) {
 		reader.Add(&res[1].buf, 100, 1000, callback2)
 		reader.Add(&res[2].buf, 0, 600, callback3)
 		reader.Add(&res[3].buf, 36, 999, callback4)
+
+		if err := reader.Error(); err != nil {
+			t.Fatalf("expected valid reader, got reader.Error: %v", err)
+		}
+
 		reader.Wait()
 		for _, k := range res {
 			if !bytes.Equal(k.buf.Bytes(), content[k.offset:k.offset+k.limit]) {
@@ -1472,6 +1477,10 @@ func TestMultiRangeDownloaderEmulated(t *testing.T) {
 		if err = reader.Close(); err != nil {
 			t.Errorf("Error while closing reader %v", err)
 		}
+
+		if err := reader.Error(); err == nil {
+			t.Fatalf("reader.Error: expected a non-nil error, got %v", err)
+		}
 	})
 }
 

storage/grpc_client.go

@@ -1500,6 +1500,15 @@ func (mrr *gRPCBidiReader) getHandle() []byte {
 	return mrr.readHandle
 }
 
+func (mrr *gRPCBidiReader) error() error {
+	mrr.mu.Lock()
+	defer mrr.mu.Unlock()
+	if mrr.done {
+		return errors.New("storage: stream is permanently closed")
+	}
+	return nil
+}
+
 func (c *grpcStorageClient) NewRangeReader(ctx context.Context, params *newRangeReaderParams, opts ...storageOption) (r *Reader, err error) {
 	// If bidi reads was not selected, use the legacy read object API.
 	if !c.config.grpcBidiReads {
@@ -1928,7 +1937,7 @@ type gRPCBidiReader struct {
 	receiverRetry    chan bool
 	mu               sync.Mutex          // protects all vars in gRPCBidiReader from concurrent access
 	mp               map[int64]rangeSpec // always use the mutex when accessing the map
-	done             bool                // always use the mutex when accessing this variable
+	done             bool                // always use the mutex when accessing this variable, indicates whether stream is closed or not.
 	activeTask       int64               // always use the mutex when accessing this variable
 	objectSize       int64               // always use the mutex when accessing this variable
 	retrier          func(error, string)

storage/reader.go

@@ -393,6 +393,7 @@ type multiRangeDownloader interface {
 	wait()
 	close() error
 	getHandle() []byte
+	error() error
 }
 
 // Add adds a new range to MultiRangeDownloader.
@@ -443,3 +444,10 @@ func (mrd *MultiRangeDownloader) Wait() {
 func (mrd *MultiRangeDownloader) GetHandle() []byte {
 	return mrd.reader.getHandle()
 }
+
+// Error returns an error if the MultiRangeDownloader is in a permanent failure
+// state. It returns a nil error if the MultiRangeDownloader is open and can be
+// used.
+func (mrd *MultiRangeDownloader) Error() error {
+	return mrd.reader.error()
+}