update multipart upload progress using reader

Author: LeMikaelF

internal/turso/tursoServer.go

@@ -21,6 +21,7 @@ type progressReader struct {
 	reader     io.Reader
 	totalSize  int64
 	bytesRead  int64
+	baseBytes  int64 // Bytes already uploaded before progressReader started
 	startTime  time.Time
 	onProgress func(progressPct int, uploadedBytes int64, totalBytes int64, elapsedTime time.Duration, done bool)
 	lastUpdate int // Last reported progress percentage. Initially -1 to ensure first update is always sent.
@@ -30,13 +31,14 @@ func (pr *progressReader) Read(p []byte) (int, error) {
 	n, err := pr.reader.Read(p)
 	if n > 0 {
 		pr.bytesRead += int64(n)
-		progressPct := int(float64(pr.bytesRead) / float64(pr.totalSize) * 100)
+		totalUploaded := pr.baseBytes + pr.bytesRead
+		progressPct := int(float64(totalUploaded) / float64(pr.totalSize) * 100)
 
 		// Only call progress if we've made at least 1% progress or if we're done
 		if progressPct > pr.lastUpdate || errors.Is(err, io.EOF) {
 			elapsedTime := time.Since(pr.startTime)
 			pr.lastUpdate = progressPct
-			pr.onProgress(progressPct, pr.bytesRead, pr.totalSize, elapsedTime, errors.Is(err, io.EOF))
+			pr.onProgress(progressPct, totalUploaded, pr.totalSize, elapsedTime, errors.Is(err, io.EOF))
 		}
 	}
 	return n, err
@@ -185,6 +187,7 @@ func (i *TursoServerClient) startMultipartUpload(dbSize int64) (int64, error) {
 func (i *TursoServerClient) uploadChunks(chunkSize int64, file io.Reader, totalSize int64, startTime time.Time, remoteEncryptionCipher, remoteEncryptionKey string, onUploadProgress func(progressPct int, uploadedBytes int64, totalBytes int64, elapsedTime time.Duration, done bool)) (int64, error) {
 	var uploadedBytes int64 = 0
 	chunkID := 0
+	lastProgressPct := -1
 
 	for uploadedBytes < totalSize {
 		remaining := totalSize - uploadedBytes
@@ -194,6 +197,16 @@ func (i *TursoServerClient) uploadChunks(chunkSize int64, file io.Reader, totalS
 		}
 
 		chunkReader := io.LimitReader(file, currentChunkSize)
+
+		progressTracker := &progressReader{
+			reader:     chunkReader,
+			totalSize:  totalSize,
+			baseBytes:  uploadedBytes,
+			startTime:  startTime,
+			onProgress: onUploadProgress,
+			lastUpdate: lastProgressPct,
+		}
+
 		chunkPath := fmt.Sprintf("/v2/upload/chunk/%d", chunkID)
 
 		var headers = map[string]string{}
@@ -203,7 +216,7 @@ func (i *TursoServerClient) uploadChunks(chunkSize int64, file io.Reader, totalS
 		}
 		headers["Content-Length"] = strconv.FormatInt(currentChunkSize, 10)
 
-		r, err := i.client.PutBinary(chunkPath, chunkReader, headers)
+		r, err := i.client.PutBinary(chunkPath, progressTracker, headers)
 		if err != nil {
 			return 0, fmt.Errorf("failed to upload chunk %d: %w", chunkID, err)
 		}
@@ -221,10 +234,7 @@ func (i *TursoServerClient) uploadChunks(chunkSize int64, file io.Reader, totalS
 		}
 
 		uploadedBytes += currentChunkSize
-		progressPct := int(float64(uploadedBytes) / float64(totalSize) * 100)
-		elapsedTime := time.Since(startTime)
-		//TODO update progress more smoothly than after every chunk
-		onUploadProgress(progressPct, uploadedBytes, totalSize, elapsedTime, false)
+		lastProgressPct = progressTracker.lastUpdate
 
 		chunkID++
 	}

internal/turso/tursoServer_test.go

@@ -175,7 +175,6 @@ func createTestFileWithContent(t *testing.T, content []byte) string {
 func createTestFile(t *testing.T, size int64) string {
 	t.Helper()
 
-	// Generate deterministic data pattern
 	pattern := []byte("TESTDATA")
 	content := make([]byte, size)
 	for i := int64(0); i < size; i++ {
@@ -926,3 +925,131 @@ func TestProgressReader_BytesReadAccurate(t *testing.T) {
 
 	require.Equal(t, int64(500), lastUploadedBytes)
 }
+
+func TestProgressReader_WithBaseBytes(t *testing.T) {
+	// Simulate reading from a second chunk of a 200-byte total upload
+	// where the first chunk (100 bytes) has already been uploaded.
+	data := strings.Repeat("x", 100) // Second chunk: 100 bytes
+	reader := strings.NewReader(data)
+
+	var progressUpdates []struct {
+		pct      int
+		uploaded int64
+		total    int64
+	}
+
+	pr := &progressReader{
+		reader:    reader,
+		totalSize: 200,
+		baseBytes: 100,
+		startTime: time.Now(),
+		onProgress: func(progressPct int, uploadedBytes int64, totalBytes int64, elapsedTime time.Duration, done bool) {
+			progressUpdates = append(progressUpdates, struct {
+				pct      int
+				uploaded int64
+				total    int64
+			}{progressPct, uploadedBytes, totalBytes})
+		},
+		lastUpdate: 50,
+	}
+
+	buf := make([]byte, 10)
+	for {
+		_, err := pr.Read(buf)
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+	}
+
+	require.NotEmpty(t, progressUpdates, "expected progress updates")
+
+	firstUpdate := progressUpdates[0]
+	require.Greater(t, firstUpdate.pct, 50, "first progress update should be > 50%%")
+
+	for i, update := range progressUpdates {
+		require.Greater(t, update.uploaded, int64(100), "update %d: uploadedBytes should be > 100 (baseBytes)", i)
+		require.Equal(t, int64(200), update.total, "update %d: totalBytes should be 200", i)
+	}
+
+	lastUpdate := progressUpdates[len(progressUpdates)-1]
+	require.Equal(t, 100, lastUpdate.pct, "final progress should be 100%%")
+	require.Equal(t, int64(200), lastUpdate.uploaded, "final uploadedBytes should be 200")
+}
+
+func TestProgressReader_CumulativeAcrossChunks(t *testing.T) {
+	totalSize := int64(300)
+	chunkSize := int64(100)
+
+	var allUpdates []struct {
+		pct      int
+		uploaded int64
+	}
+
+	// Simulate reading 3 chunks
+	var baseBytes int64 = 0
+	lastPct := -1
+
+	for chunk := 0; chunk < 3; chunk++ {
+		data := bytes.Repeat([]byte("x"), int(chunkSize))
+		reader := bytes.NewReader(data)
+
+		pr := &progressReader{
+			reader:    reader,
+			totalSize: totalSize,
+			baseBytes: baseBytes,
+			startTime: time.Now(),
+			onProgress: func(progressPct int, uploadedBytes int64, totalBytes int64, elapsedTime time.Duration, done bool) {
+				allUpdates = append(allUpdates, struct {
+					pct      int
+					uploaded int64
+				}{progressPct, uploadedBytes})
+			},
+			lastUpdate: lastPct,
+		}
+
+		// Read all of this chunk
+		buf := make([]byte, 10)
+		for {
+			_, err := pr.Read(buf)
+			if err == io.EOF {
+				break
+			}
+			require.NoError(t, err)
+		}
+
+		baseBytes += chunkSize
+		lastPct = pr.lastUpdate
+	}
+
+	require.NotEmpty(t, allUpdates, "expected progress updates")
+	require.LessOrEqual(t, allUpdates[0].pct, 10, "first update should be low")
+	require.Equal(t, 100, allUpdates[len(allUpdates)-1].pct, "last update should be 100%%")
+	for i := 1; i < len(allUpdates); i++ {
+		require.GreaterOrEqual(t, allUpdates[i].pct, allUpdates[i-1].pct, "progress went backwards: %d%% -> %d%%", allUpdates[i-1].pct, allUpdates[i].pct)
+	}
+	for i := 1; i < len(allUpdates); i++ {
+		require.GreaterOrEqual(t, allUpdates[i].uploaded, allUpdates[i-1].uploaded, "uploadedBytes went backwards: %d -> %d", allUpdates[i-1].uploaded, allUpdates[i].uploaded)
+	}
+}
+
+func TestUploadFileMultipart_SmoothProgress(t *testing.T) {
+	mock := NewMockTursoServer()
+	mock.chunkSize = 100 * 1024
+	defer mock.Close()
+
+	client := createTestClient(t, mock.URL)
+	testFile := createTestFile(t, 1024*1024)
+	progress := NewProgressRecorder()
+
+	err := client.UploadFileMultipart(testFile, "", "", progress.Callback())
+	require.NoError(t, err)
+
+	calls := progress.GetCalls()
+
+	require.Greater(t, len(calls), 15, "Expected smooth progress with many updates, got only %d", len(calls))
+	progress.VerifyProgressIncreasing(t)
+	require.True(t, calls[len(calls)-1].Done, "Final callback should have Done=true")
+}