[qob] maybe retry GSFS reads (hail-is#13013)

Fixes hail-is#12983

---

After an `FSSeekableInputStream` method (successfully) returns,
`getPosition` always represents theintended location within the object.
We can entirely eliminate `lazyPosition` because it tracks the
same value as `getPosition` when `reader == null`.

For retryable reads, we just seek back to the known correct location of
the stream before attempting to read again.

hail/src/main/scala/is/hail/io/fs/GoogleStorageFS.scala

@@ -198,32 +198,30 @@ class GoogleStorageFS(
 
     val is: SeekableInputStream = new FSSeekableInputStream {
       private[this] var reader: ReadChannel = null
-      private[this] var lazyPosition: Long = 0L
+
+      private[this] def retryingRead(): Int = {
+        retryTransientErrors(
+          { reader.read(bb) },
+          reset = Some({ () => reader.seek(getPosition) })
+        )
+      }
 
       private[this] def readHandlingRequesterPays(bb: ByteBuffer): Int = {
         if (reader != null) {
-          reader.read(bb)
+          retryingRead()
         } else {
           handleRequesterPays(
             { (options: Seq[BlobSourceOption]) =>
               reader = retryTransientErrors { storage.reader(bucket, path, options:_*) }
-              reader.seek(lazyPosition)
-              reader.read(bb)
+              reader.seek(getPosition)
+              retryingRead()
             },
             BlobSourceOption.userProject _,
             bucket
           )
         }
       }
 
-      private[this] def seekHandlingRequesterPays(newPos: Long): Unit = {
-        if (reader != null) {
-          reader.seek(newPos)
-        } else {
-          lazyPosition = newPos
-        }
-      }
-
       override def close(): Unit = {
         if (!closed) {
           if (reader != null) {
@@ -251,7 +249,9 @@ class GoogleStorageFS(
       }
 
       override def physicalSeek(newPos: Long): Unit = {
-        seekHandlingRequesterPays(newPos)
+        if (reader != null) {
+          reader.seek(newPos)
+        }
       }
     }
 

hail/src/main/scala/is/hail/services/package.scala

@@ -116,7 +116,7 @@ package object services {
     }
   }
 
-  def retryTransientErrors[T](f: => T): T = {
+  def retryTransientErrors[T](f: => T, reset: Option[() => Unit] = None): T = {
     var delay = 0.1
     var errors = 0
     while (true) {
@@ -137,6 +137,7 @@ package object services {
           }
       }
       delay = sleepAndBackoff(delay)
+      reset.foreach(_())
     }
 
     throw new AssertionError("unreachable")