Merge branch '10700-dispatch'

[arvados.git] / services / keep-balance / collection.go

diff --git a/services/keep-balance/collection.go b/services/keep-balance/collection.go
index e8ac30bd7acea506508cbdd0d9036d763aac89d3..a7a484672a0c3c073717ac1e65e77c44af21cdf5 100644 (file)
--- a/services/keep-balance/collection.go
+++ b/services/keep-balance/collection.go
@@ -53,6 +53,7 @@ func EachCollection(c *arvados.Client, pageSize int, f func(arvados.Collection)
        var last arvados.Collection
        var filterTime time.Time
        callCount := 0
+       gettingExactTimestamp := false
        for {
                progress(callCount, expectCount)
                var page arvados.CollectionList
@@ -71,26 +72,62 @@ func EachCollection(c *arvados.Client, pageSize int, f func(arvados.Collection)
                        }
                        last = coll
                }
-               if last.ModifiedAt == nil || *last.ModifiedAt == filterTime {
-                       if page.ItemsAvailable > len(page.Items) {
-                               // TODO: use "mtime=X && UUID>Y"
-                               // filters to get all collections with
-                               // this timestamp, then use "mtime>X"
-                               // to get the next timestamp.
-                               return fmt.Errorf("BUG: Received an entire page with the same modified_at timestamp (%v), cannot make progress", filterTime)
-                       }
+               if len(page.Items) == 0 && !gettingExactTimestamp {
                        break
+               } else if last.ModifiedAt == nil {
+                       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 {
+                       // 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
+                       // gettingExactTimestamp mode (if we're not
+                       // there already), advancing our UUID
+                       // threshold with each request, until we get
+                       // an empty page.
+                       gettingExactTimestamp = true
+                       params.Filters = []arvados.Filter{{
+                               Attr:     "modified_at",
+                               Operator: "=",
+                               Operand:  filterTime,
+                       }, {
+                               Attr:     "uuid",
+                               Operator: ">",
+                               Operand:  last.UUID,
+                       }}
+               } else if gettingExactTimestamp {
+                       // This must be an empty page (in this mode,
+                       // an unequal timestamp is impossible) so we
+                       // can start getting pages of newer
+                       // collections.
+                       gettingExactTimestamp = false
+                       params.Filters = []arvados.Filter{{
+                               Attr:     "modified_at",
+                               Operator: ">",
+                               Operand:  filterTime,
+                       }}
+               } else {
+                       // In the normal case, we know we have seen
+                       // all collections with modtime<filterTime,
+                       // but we might not have seen all that have
+                       // modtime=filterTime. Hence we use >= instead
+                       // of > and skip the obvious overlapping item,
+                       // i.e., the last item on the previous
+                       // page. In some edge cases this can return
+                       // collections we have already seen, but
+                       // avoiding that would add overhead in the
+                       // overwhelmingly common cases, so we don't
+                       // bother.
+                       filterTime = *last.ModifiedAt
+                       params.Filters = []arvados.Filter{{
+                               Attr:     "modified_at",
+                               Operator: ">=",
+                               Operand:  filterTime,
+                       }, {
+                               Attr:     "uuid",
+                               Operator: "!=",
+                               Operand:  last.UUID,
+                       }}
                }
-               filterTime = *last.ModifiedAt
-               params.Filters = []arvados.Filter{{
-                       Attr:     "modified_at",
-                       Operator: ">=",
-                       Operand:  filterTime,
-               }, {
-                       Attr:     "uuid",
-                       Operator: "!=",
-                       Operand:  last.UUID,
-               }}
        }
        progress(callCount, expectCount)