Change workqueue to prefer lowest timestamp. (#1160)

Currently we choose the highest timestamp, which inhibits our ability to
force trying a key earlier.

If we have a low priority key queued with a long delay, and we want to
try it with high priority now, the result is a high priority with the
long delay timestamp.

In #1159 I
want to be able to periodically resync N keys by loading the key space
and queuing all of the keys in the future, but this may be hours in the
future.

When a key actually changes, we want to be able to process it
immediately with high priority, but the current semantics mean that if
there is an outstanding resync that some keys will get starved until the
resync queued keys are processed.

```
ok      github.com/chainguard-dev/terraform-infra-common/pkg/workqueue/gcs      316.861s
```

Signed-off-by: Matt Moore <[email protected]>

Author: mattmoor

pkg/workqueue/conformance/conformance.go

@@ -361,21 +361,8 @@ func TestSemantics(t *testing.T, ctor func(int) workqueue.Interface) {
 			t.Fatalf("Start failed: %v", err)
 		}
 
-		// Queue the in-progress key with no priority.
-		if err := wq.Queue(ctx, owned.Name(), workqueue.Options{
-			// No priority
-		}); err != nil {
-			t.Fatalf("Queue failed: %v", err)
-		}
-
-		// Check that the key is queued and in-progress, and the queued key
-		// is behind the other key.
-		_, _ = checkQueue(t, wq, ExpectedState{
-			WorkInProgress: []string{"foo"},
-			Queued:         []string{"bar", "foo"},
-		})
-
 		// Requeue the in-progress high-priority key.
+		// This will add backoff NotBefore.
 		if err := owned.Requeue(ctx); err != nil {
 			t.Fatalf("Requeue failed: %v", err)
 		}
@@ -422,30 +409,7 @@ func TestSemantics(t *testing.T, ctor func(int) workqueue.Interface) {
 			t.Fatalf("Queue failed: %v", err)
 		}
 
-		// The queue should appear empty because the later NotBefore won.
-		_, _ = checkQueue(t, wq, ExpectedState{})
-
-		// Queue the same key again with a NotBefore that's twice as long.
-		if err := wq.Queue(ctx, "foo", workqueue.Options{
-			NotBefore: time.Now().UTC().Add(2 * delay),
-		}); err != nil {
-			t.Fatalf("Queue failed: %v", err)
-		}
-
-		// The queue should appear empty
-		_, _ = checkQueue(t, wq, ExpectedState{})
-
-		// Sleep for the NotBefore delay.
-		time.Sleep(delay)
-
-		// The queue should STILL appear empty because the doubled delay
-		// should have won.
-		_, _ = checkQueue(t, wq, ExpectedState{})
-
-		// Sleep for the NotBefore delay one last time.
-		time.Sleep(delay)
-
-		// The queue should now have the key.
+		// The queue should still have the key because the earlier NotBefore won.
 		_, _ = checkQueue(t, wq, ExpectedState{
 			Queued: []string{"foo"},
 		})
@@ -487,53 +451,33 @@ func TestSemantics(t *testing.T, ctor func(int) workqueue.Interface) {
 	})
 
 	ct.scenario("requeue doesn't reset not before", func(ctx context.Context, t *testing.T, wq workqueue.Interface) {
-		// Queue a key without NotBefore set.
+		// Test that lowest (earliest) NotBefore wins when merging.
+		// Queue a key with a long NotBefore delay.
+		longDelay := 10 * workqueue.BackoffPeriod
 		if err := wq.Queue(ctx, "foo", workqueue.Options{
-			// No NotBefore.
+			NotBefore: time.Now().UTC().Add(longDelay),
 		}); err != nil {
 			t.Fatalf("Queue failed: %v", err)
 		}
 
-		// The queue should have the key.
-		_, qd := checkQueue(t, wq, ExpectedState{
-			Queued: []string{"foo"},
-		})
-
-		// Start processing the first key.
-		owned, err := qd[0].Start(ctx)
-		if err != nil {
-			t.Fatalf("Start failed: %v", err)
-		}
-
-		// The key should now be in progress
-		_, _ = checkQueue(t, wq, ExpectedState{
-			WorkInProgress: []string{"foo"},
-		})
+		// The queue should be empty (long delay not passed).
+		_, _ = checkQueue(t, wq, ExpectedState{})
 
-		// Queue the key again with a short NotBefore delay.
-		if err := wq.Queue(ctx, owned.Name(), workqueue.Options{
-			NotBefore: time.Now().UTC().Add(delay),
+		// Queue the same key again with a shorter NotBefore delay.
+		shortDelay := workqueue.BackoffPeriod
+		if err := wq.Queue(ctx, "foo", workqueue.Options{
+			NotBefore: time.Now().UTC().Add(shortDelay),
 		}); err != nil {
 			t.Fatalf("Queue failed: %v", err)
 		}
 
-		// The queue should still have the key just in-progress.
-		_, _ = checkQueue(t, wq, ExpectedState{
-			WorkInProgress: []string{"foo"},
-		})
-
-		// Requeue the key.
-		if err := owned.Requeue(ctx); err != nil {
-			t.Fatalf("Requeue failed: %v", err)
-		}
-
-		// The requeue should not reset NotBefore.
+		// The queue should still be empty (short delay not passed yet).
 		_, _ = checkQueue(t, wq, ExpectedState{})
 
-		// Sleep for the NotBefore delay.
-		time.Sleep(delay)
+		// Sleep for the short delay.
+		time.Sleep(shortDelay)
 
-		// Now the key should show as queued.
+		// Now the key should show as queued (short delay won over long delay).
 		_, _ = checkQueue(t, wq, ExpectedState{
 			Queued: []string{"foo"},
 		})

pkg/workqueue/gcs/gcs.go

@@ -117,23 +117,6 @@ func (w *wq) Queue(ctx context.Context, key string, opts workqueue.Options) erro
 }
 
 func updateMetadata(ctx context.Context, client ClientInterface, key string, metadata map[string]string) error {
-	switch {
-	case metadata[priorityMetadataKey] != noPriority:
-		// If the priority was set, then attempt to merge it with the queued
-		// key.
-		break
-
-	case metadata[notBeforeMetadataKey] != noNotBefore:
-		// If the not before was set, then attempt to merge it with the queued
-		// key.  This is largely for Queue operations, as the NotBefore is
-		// cleared when we start processing the key.
-		break
-
-	default:
-		// No options, so don't bother fetching the queued object.
-		return nil
-	}
-
 	attrs, err := client.Object(fmt.Sprintf("%s%s", queuedPrefix, key)).Attrs(ctx)
 	if err != nil {
 		return fmt.Errorf("Attrs() = %w", err)
@@ -143,12 +126,14 @@ func updateMetadata(ctx context.Context, client ClientInterface, key string, met
 		attrs.Metadata = make(map[string]string, 2)
 	}
 	update := false
+	// Always choose the highest priority.
 	if p, ok := attrs.Metadata[priorityMetadataKey]; !ok || p < metadata[priorityMetadataKey] {
 		clog.InfoContextf(ctx, "Updating %s priority from %q to %q", key, p, metadata[priorityMetadataKey])
 		attrs.Metadata[priorityMetadataKey] = metadata[priorityMetadataKey]
 		update = true
 	}
-	if ts, ok := attrs.Metadata[notBeforeMetadataKey]; ok && ts < metadata[notBeforeMetadataKey] {
+	// Always choose the lowest not-before.
+	if ts, ok := attrs.Metadata[notBeforeMetadataKey]; ok && ts > metadata[notBeforeMetadataKey] {
 		clog.InfoContextf(ctx, "Updating %s not-before from %q to %q", key, ts, metadata[notBeforeMetadataKey])
 		attrs.Metadata[notBeforeMetadataKey] = metadata[notBeforeMetadataKey]
 		update = true

pkg/workqueue/inmem/inmem.go

@@ -49,16 +49,21 @@ var _ workqueue.Interface = (*wq)(nil)
 func (w *wq) Queue(_ context.Context, key string, opts workqueue.Options) error {
 	w.rw.Lock()
 	defer w.rw.Unlock()
-	if qi, ok := w.queue[key]; !ok {
+	qi, ok := w.queue[key]
+	if !ok {
 		w.queue[key] = queueItem{
 			Options: opts,
 			queued:  time.Now().UTC(),
 		}
-	} else if qi.Priority < opts.Priority {
-		// Raise the priority of the queued item.
+		return nil
+	}
+	// Always choose the highest priority.
+	if qi.Priority < opts.Priority {
 		qi.Priority = opts.Priority
 		w.queue[key] = qi
-	} else if qi.NotBefore.Before(opts.NotBefore) {
+	}
+	// Always choose the lowest not-before.
+	if qi.NotBefore.After(opts.NotBefore) {
 		// Update the NotBefore time.
 		qi.NotBefore = opts.NotBefore
 		w.queue[key] = qi
@@ -223,12 +228,12 @@ func (o *inProgressKey) RequeueWithOptions(_ context.Context, opts workqueue.Opt
 			queued:   time.Now().UTC(),
 		}
 	} else {
+		// Always choose the highest priority.
 		if qi.Priority < opts.Priority {
-			// Raise the priority of the queued item.
 			qi.Priority = opts.Priority
 		}
-		if opts.NotBefore.After(qi.NotBefore) {
-			// Update the NotBefore time if the new one is later.
+		// Always choose the lowest not-before.
+		if qi.NotBefore.After(opts.NotBefore) {
 			qi.NotBefore = opts.NotBefore
 		}
 		o.wq.queue[o.key] = qi