Fix WebDAV upload failing when the pooled connection was closed

buildsystem/dependencies.gradle

@@ -182,6 +182,7 @@ ext {
 			mockitoInline                  : "org.mockito:mockito-inline:${mockitoInlineVersion}",
 			mockitoKotlin                  : "org.mockito.kotlin:mockito-kotlin:${mockitoKotlinVersion}",
 			mockitoAndroid                 : "org.mockito:mockito-android:${mockitoAndroidVersion}",
+			mockWebServer                  : "com.squareup.okhttp3:mockwebserver3:${okHttpVersion}",
 			msgraph                        : "com.microsoft.graph:microsoft-graph:${msgraphVersion}",
 			msgraphAuth                    : "com.microsoft.identity.client:msal:${msgraphAuthVersion}",
 			okHttp                         : "com.squareup.okhttp3:okhttp:${okHttpVersion}",

data/build.gradle

@@ -217,6 +217,7 @@ dependencies {
 	implementation dependencies.androidxTestJunitKtln
 
 	testImplementation dependencies.mockito
+	testImplementation dependencies.mockWebServer
 	testImplementation dependencies.mockitoKotlin
 	testImplementation dependencies.mockitoInline
 	testImplementation dependencies.hamcrest

data/src/main/java/org/cryptomator/data/cloud/webdav/WebDavImpl.kt

@@ -2,6 +2,7 @@ package org.cryptomator.data.cloud.webdav
 
 import android.content.Context
 import org.cryptomator.data.cloud.webdav.network.ConnectionHandlerHandlerImpl
+import org.cryptomator.data.cloud.webdav.network.DataSourceBasedRequestBody
 import org.cryptomator.data.cloud.webdav.network.ServerNotWebdavCompatibleException
 import org.cryptomator.data.util.CopyStream
 import org.cryptomator.data.util.TransferredBytesAwareInputStream
@@ -123,7 +124,7 @@ internal class WebDavImpl(private val cloud: WebDavCloud, private val connection
 		}
 
 		progressAware.onProgress(Progress.started(UploadState.upload(uploadFile)))
-		data.open(context)?.use { inputStream ->
+		val requestBody = DataSourceBasedRequestBody.from(context, data, size) { inputStream ->
 			object : TransferredBytesAwareInputStream(inputStream) {
 				override fun bytesTransferred(transferred: Long) {
 					progressAware.onProgress( //
@@ -133,10 +134,9 @@ internal class WebDavImpl(private val cloud: WebDavCloud, private val connection
 							.withValue(transferred)
 					)
 				}
-			}.use {
-				connectionHandler.writeFile(absoluteUriFrom(uploadFile.path), it, data.modifiedDate(context).orElse(Date()))
 			}
-		} ?: throw FatalBackendException("InputStream shouldn't bee null")
+		}
+		connectionHandler.writeFile(absoluteUriFrom(uploadFile.path), requestBody, data.modifiedDate(context).orElse(Date()))
 
 		return connectionHandler.get(absoluteUriFrom(uploadFile.path), uploadFile.parent) as WebDavFile? ?: throw FatalBackendException("Unable to get CloudFile after upload.")
 	}

data/src/main/java/org/cryptomator/data/cloud/webdav/network/ConnectionHandlerHandlerImpl.kt

@@ -7,6 +7,7 @@ import org.cryptomator.domain.exception.BackendException
 import java.io.InputStream
 import java.util.Date
 import javax.inject.Inject
+import okhttp3.RequestBody
 
 class ConnectionHandlerHandlerImpl @Inject internal constructor(httpClient: WebDavCompatibleHttpClient) {
 
@@ -28,8 +29,8 @@ class ConnectionHandlerHandlerImpl @Inject internal constructor(httpClient: WebD
 	}
 
 	@Throws(BackendException::class)
-	fun writeFile(url: String, inputStream: InputStream, modifiedDate: Date) {
-		webDavClient.writeFile(url, inputStream, modifiedDate)
+	fun writeFile(url: String, requestBody: RequestBody, modifiedDate: Date) {
+		webDavClient.writeFile(url, requestBody, modifiedDate)
 	}
 
 	@Throws(BackendException::class)

data/src/main/java/org/cryptomator/data/cloud/webdav/network/DataSourceBasedRequestBody.kt

@@ -0,0 +1,52 @@
+package org.cryptomator.data.cloud.webdav.network
+
+import android.content.Context
+import org.cryptomator.domain.exception.FatalBackendException
+import org.cryptomator.domain.usecases.cloud.DataSource
+import java.io.IOException
+import java.io.InputStream
+import okhttp3.MediaType
+import okhttp3.MediaType.Companion.toMediaTypeOrNull
+import okhttp3.RequestBody
+import okio.BufferedSink
+import okio.source
+
+
+internal class DataSourceBasedRequestBody private constructor( //
+	private val context: Context,  //
+	private val data: DataSource,  //
+	private val size: Long,  //
+	private val decorate: (InputStream) -> InputStream
+) : RequestBody() {
+
+	override fun contentLength(): Long {
+		return size
+	}
+
+	override fun contentType(): MediaType? {
+		return "application/octet-stream".toMediaTypeOrNull()
+	}
+
+	/**
+	 * Opens the data again on every invocation instead of consuming a single stream. Without that, OkHttp is unable to
+	 * repeat the request after it used a pooled connection the server closed in the meantime, e.g. because of its
+	 * keep-alive timeout, and the upload fails instead of being retried using a new connection.
+	 * see https://github.com/cryptomator/android/issues/646
+	 */
+	@Throws(IOException::class)
+	override fun writeTo(sink: BufferedSink) {
+		data.open(context)?.use { inputStream ->
+			decorate(inputStream).source().use {
+				sink.writeAll(it)
+			}
+		} ?: throw FatalBackendException("InputStream shouldn't be null")
+	}
+
+	companion object {
+
+		fun from(context: Context, data: DataSource, size: Long, decorate: (InputStream) -> InputStream): RequestBody {
+			return DataSourceBasedRequestBody(context, data, size, decorate)
+		}
+
+	}
+}

data/src/main/java/org/cryptomator/data/cloud/webdav/network/WebDavClient.kt

@@ -21,6 +21,7 @@ import java.util.Collections
 import java.util.Date
 import okhttp3.MediaType.Companion.toMediaTypeOrNull
 import okhttp3.Request
+import okhttp3.RequestBody
 import okhttp3.RequestBody.Companion.toRequestBody
 import okhttp3.Response
 
@@ -151,10 +152,10 @@ internal class WebDavClient(private val httpClient: WebDavCompatibleHttpClient)
 	}
 
 	@Throws(BackendException::class)
-	fun writeFile(url: String, inputStream: InputStream, modifiedDate: Date) {
+	fun writeFile(url: String, requestBody: RequestBody, modifiedDate: Date) {
 		val builder = Request.Builder() //
 			.addHeader("X-OC-Mtime", modifiedDate.toInstant().toEpochMilli().div(1000).toString()) //
-			.put(InputStreamSourceBasedRequestBody.from(inputStream)) //
+			.put(requestBody) //
 			.url(url)
 		try {
 			httpClient.execute(builder).use { response ->

data/src/test/java/org/cryptomator/data/cloud/webdav/network/DataSourceBasedRequestBodyRetryTest.kt

@@ -0,0 +1,83 @@
+package org.cryptomator.data.cloud.webdav.network
+
+import android.content.Context
+import org.cryptomator.domain.usecases.cloud.ByteArrayDataSource
+import org.hamcrest.CoreMatchers
+import org.hamcrest.MatcherAssert
+import org.junit.jupiter.api.AfterEach
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.DisplayName
+import org.junit.jupiter.api.Test
+import org.mockito.kotlin.mock
+import java.nio.charset.StandardCharsets
+import mockwebserver3.MockResponse
+import mockwebserver3.MockWebServer
+import mockwebserver3.RecordedRequest
+import mockwebserver3.SocketEffect
+import okhttp3.OkHttpClient
+import okhttp3.Request
+
+class DataSourceBasedRequestBodyRetryTest {
+
+	private val context = mock<Context>()
+
+	private lateinit var server: MockWebServer
+
+	@BeforeEach
+	fun setup() {
+		server = MockWebServer()
+		server.start()
+	}
+
+	@AfterEach
+	fun tearDown() {
+		server.close()
+	}
+
+	/**
+	 * Reproduces the upload failing after the server closed the connection it was pooled on, e.g. because of the
+	 * keep-alive timeout of Apache, which defaults to five seconds. Opening a text file downloads it and leaves the
+	 * connection in the pool, saving it reuses that connection. OkHttp repeats such a request using a new connection,
+	 * which only succeeds if the request body writes the complete content again.
+	 * see https://github.com/cryptomator/android/issues/646
+	 */
+	@Test
+	@DisplayName("upload is repeated with the complete content after the server closed the pooled connection")
+	fun testUploadIsRepeatedWithTheCompleteContentAfterTheServerClosedThePooledConnection() {
+		server.enqueue(MockResponse.Builder().code(200).body("Wer die Wahl hat").build())
+		server.enqueue(MockResponse.Builder().onResponseStart(SocketEffect.CloseSocket(true, true, true)).build())
+		server.enqueue(MockResponse.Builder().code(204).build())
+
+		val url = server.url("/vault/d/AB/CDEFGH.c9r")
+		val client = OkHttpClient()
+
+		client.newCall(Request.Builder().url(url).build()).execute().use { response ->
+			MatcherAssert.assertThat(response.code, CoreMatchers.`is`(200))
+		}
+
+		val requestBody = DataSourceBasedRequestBody.from(context, ByteArrayDataSource.from(CONTENT), CONTENT.size.toLong()) { it }
+		val request = Request.Builder() //
+			.put(requestBody) //
+			.url(url) //
+			.build()
+
+		client.newCall(request).execute().use { response ->
+			MatcherAssert.assertThat(response.code, CoreMatchers.`is`(204))
+		}
+
+		val uploads = recordedRequests().filter { it.method == "PUT" }
+		MatcherAssert.assertThat(uploads.size, CoreMatchers.`is`(2))
+		MatcherAssert.assertThat(uploads.last().body?.toByteArray(), CoreMatchers.`is`(CONTENT))
+		MatcherAssert.assertThat(uploads.last().connectionIndex, CoreMatchers.not(uploads.first().connectionIndex))
+	}
+
+	private fun recordedRequests(): List<RecordedRequest> {
+		return (0 until server.requestCount).map { server.takeRequest() }
+	}
+
+	companion object {
+
+		private val CONTENT = "Wer die Wahl hat, hat die Qual".toByteArray(StandardCharsets.UTF_8)
+
+	}
+}

data/src/test/java/org/cryptomator/data/cloud/webdav/network/DataSourceBasedRequestBodyTest.kt

@@ -0,0 +1,117 @@
+package org.cryptomator.data.cloud.webdav.network
+
+import android.content.Context
+import org.cryptomator.data.util.TransferredBytesAwareInputStream
+import org.cryptomator.domain.usecases.cloud.ByteArrayDataSource
+import org.cryptomator.domain.usecases.cloud.DataSource
+import org.cryptomator.util.Optional
+import org.hamcrest.CoreMatchers
+import org.hamcrest.MatcherAssert
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.DisplayName
+import org.junit.jupiter.api.Test
+import org.mockito.kotlin.mock
+import java.io.InputStream
+import java.nio.charset.StandardCharsets
+import java.util.Date
+import okio.Buffer
+
+class DataSourceBasedRequestBodyTest {
+
+	private val context = mock<Context>()
+
+	private lateinit var data: CountingDataSource
+
+	@BeforeEach
+	fun setup() {
+		data = CountingDataSource(ByteArrayDataSource.from(CONTENT))
+	}
+
+	@Test
+	@DisplayName("contentLength() returns the size the request body was created with")
+	fun testContentLengthReturnsTheSizeTheRequestBodyWasCreatedWith() {
+		val inTest = DataSourceBasedRequestBody.from(context, data, CONTENT.size.toLong()) { it }
+
+		MatcherAssert.assertThat(inTest.contentLength(), CoreMatchers.`is`(CONTENT.size.toLong()))
+	}
+
+	@Test
+	@DisplayName("isOneShot() is false because the request body can be written more than once")
+	fun testIsOneShotIsFalse() {
+		val inTest = DataSourceBasedRequestBody.from(context, data, CONTENT.size.toLong()) { it }
+
+		MatcherAssert.assertThat(inTest.isOneShot(), CoreMatchers.`is`(false))
+	}
+
+	/**
+	 * OkHttp repeats a request when it used a pooled connection the server closed in the meantime, e.g. because of its
+	 * keep-alive timeout. Writing the request body a second time has to write the complete content again instead of
+	 * failing on the stream consumed by the first attempt. see https://github.com/cryptomator/android/issues/646
+	 */
+	@Test
+	@DisplayName("writeTo(…) writes the complete content on every attempt")
+	fun testWriteToWritesTheCompleteContentOnEveryAttempt() {
+		val inTest = DataSourceBasedRequestBody.from(context, data, CONTENT.size.toLong()) { it }
+
+		val firstAttempt = Buffer()
+		inTest.writeTo(firstAttempt)
+		val secondAttempt = Buffer()
+		inTest.writeTo(secondAttempt)
+
+		MatcherAssert.assertThat(firstAttempt.readByteArray(), CoreMatchers.`is`(CONTENT))
+		MatcherAssert.assertThat(secondAttempt.readByteArray(), CoreMatchers.`is`(CONTENT))
+		MatcherAssert.assertThat(data.opened, CoreMatchers.`is`(2))
+	}
+
+	@Test
+	@DisplayName("writeTo(…) reports the transferred bytes of every attempt")
+	fun testWriteToReportsTheTransferredBytesOfEveryAttempt() {
+		val reported = ArrayList<Long>()
+		val inTest = DataSourceBasedRequestBody.from(context, data, CONTENT.size.toLong()) { inputStream ->
+			object : TransferredBytesAwareInputStream(inputStream) {
+				override fun bytesTransferred(transferred: Long) {
+					reported.add(transferred)
+				}
+			}
+		}
+
+		inTest.writeTo(Buffer())
+		reported.clear()
+		inTest.writeTo(Buffer())
+
+		MatcherAssert.assertThat(reported.lastOrNull(), CoreMatchers.`is`(CONTENT.size.toLong()))
+	}
+
+	private class CountingDataSource(private val delegate: DataSource) : DataSource {
+
+		var opened = 0
+			private set
+
+		override fun size(context: Context): Long? {
+			return delegate.size(context)
+		}
+
+		override fun open(context: Context): InputStream? {
+			opened++
+			return delegate.open(context)
+		}
+
+		override fun decorate(delegate: DataSource): DataSource {
+			return delegate
+		}
+
+		override fun close() {
+			delegate.close()
+		}
+
+		override fun modifiedDate(context: Context): Optional<Date> {
+			return delegate.modifiedDate(context)
+		}
+	}
+
+	companion object {
+
+		private val CONTENT = "Wer die Wahl hat, hat die Qual".toByteArray(StandardCharsets.UTF_8)
+
+	}
+}

domain/src/main/java/org/cryptomator/domain/usecases/cloud/DataSource.kt

@@ -12,6 +12,10 @@ interface DataSource : Serializable, Closeable {
 
 	fun size(context: Context): Long?
 
+	/**
+	 * Opens a new stream on every invocation so that the data can be read more than once, which is required whenever a
+	 * request carrying it has to be repeated. see https://github.com/cryptomator/android/issues/646
+	 */
 	@Throws(IOException::class)
 	fun open(context: Context): InputStream?