X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/2b8f39b5067103d60f7bc43d26cb2f59126b2a42..865e5c1e3730117870eb1e485d553383626b882f:/services/keep-balance/collection.go

diff --git a/services/keep-balance/collection.go b/services/keep-balance/collection.go
index 1e5fa57978..1659918caf 100644
--- a/services/keep-balance/collection.go
+++ b/services/keep-balance/collection.go
@@ -5,10 +5,11 @@
 package main
 
 import (
+	"context"
 	"fmt"
 	"time"
 
-	"git.curoverse.com/arvados.git/sdk/go/arvados"
+	"git.arvados.org/arvados.git/sdk/go/arvados"
 )
 
 func countCollections(c *arvados.Client, params arvados.ResourceListParams) (int, error) {
@@ -30,7 +31,7 @@ func countCollections(c *arvados.Client, params arvados.ResourceListParams) (int
 //
 // If pageSize > 0 it is used as the maximum page size in each API
 // call; otherwise the maximum allowed page size is requested.
-func EachCollection(c *arvados.Client, pageSize int, f func(arvados.Collection) error, progress func(done, total int)) error {
+func EachCollection(ctx context.Context, c *arvados.Client, pageSize int, f func(arvados.Collection) error, progress func(done, total int)) error {
 	if progress == nil {
 		progress = func(_, _ int) {}
 	}
@@ -43,6 +44,18 @@ func EachCollection(c *arvados.Client, pageSize int, f func(arvados.Collection)
 		return err
 	}
 
+	// Note the obvious way to get all collections (sorting by
+	// UUID) would be much easier, but would lose data: If a
+	// client were to move files from collection with uuid="zzz"
+	// to a collection with uuid="aaa" around the time when we
+	// were fetching the "mmm" page, we would never see those
+	// files' block IDs at all -- even if the client is careful to
+	// save "aaa" before saving "zzz".
+	//
+	// Instead, we get pages in modified_at order. Collections
+	// that are modified during the run will be re-fetched in a
+	// subsequent page.
+
 	limit := pageSize
 	if limit <= 0 {
 		// Use the maximum page size the server allows
@@ -63,12 +76,12 @@ func EachCollection(c *arvados.Client, pageSize int, f func(arvados.Collection)
 	for {
 		progress(callCount, expectCount)
 		var page arvados.CollectionList
-		err := c.RequestAndDecode(&page, "GET", "arvados/v1/collections", nil, params)
+		err := c.RequestAndDecodeContext(ctx, &page, "GET", "arvados/v1/collections", nil, params)
 		if err != nil {
 			return err
 		}
 		for _, coll := range page.Items {
-			if last.ModifiedAt != nil && *last.ModifiedAt == *coll.ModifiedAt && last.UUID >= coll.UUID {
+			if last.ModifiedAt == coll.ModifiedAt && last.UUID >= coll.UUID {
 				continue
 			}
 			callCount++
@@ -80,9 +93,9 @@ func EachCollection(c *arvados.Client, pageSize int, f func(arvados.Collection)
 		}
 		if len(page.Items) == 0 && !gettingExactTimestamp {
 			break
-		} else if last.ModifiedAt == nil {
+		} else if last.ModifiedAt.IsZero() {
 			return fmt.Errorf("BUG: Last collection on the page (%s) has no modified_at timestamp; cannot make progress", last.UUID)
-		} else if len(page.Items) > 0 && *last.ModifiedAt == filterTime {
+		} else if len(page.Items) > 0 && last.ModifiedAt == filterTime {
 			// If we requested time>=X and never got a
 			// time>X then we might not have received all
 			// items with time==X yet. Switch to
@@ -123,7 +136,7 @@ func EachCollection(c *arvados.Client, pageSize int, f func(arvados.Collection)
 			// avoiding that would add overhead in the
 			// overwhelmingly common cases, so we don't
 			// bother.
-			filterTime = *last.ModifiedAt
+			filterTime = last.ModifiedAt
 			params.Filters = []arvados.Filter{{
 				Attr:     "modified_at",
 				Operator: ">=",