cmd/worker: Use db.LatestIndexTimestamp for internal poller rather than time horizon.

This is my mistake, sorry. In
https://go-review.googlesource.com/c/pkgsite/+/747880 I added the
internal poller with a horizon option, thinking that that was the
simplest way to do this. But, the problem with the horizon is that the
"cursor" doesn't advance properly. It advances with time, not with
whatever we indexed last.

If we last indexed T1, and the worker dies, we'd like to resume from T1.
Instead, with the horizon, we resume from whatever `Tx=now()-horizon`
might be. That's different than how /poll works: it uses
db.LatestIndexTimestamp. So, it'd have resumed from T1, since that was
where we last indexed. This change basically just makes the internal
poller do the same thing as the /poll endpoint in that regard.

Change-Id: I3232f1684d29ffc0c3bd5104fcdb8562899a2b93
Reviewed-on: https://go-review.googlesource.com/c/pkgsite/+/803720
Auto-Submit: Jonathan Amsterdam <jba@google.com>
Auto-Submit: Ethan Lee <ethanalee@google.com>
Reviewed-by: Jonathan Amsterdam <jba@google.com>
kokoro-CI: kokoro <noreply+kokoro@google.com>
Reviewed-by: Ethan Lee <ethanalee@google.com>
Commit-Queue: Jonathan Amsterdam <jba@google.com>
LUCI-TryBot-Result: golang-scoped@luci-project-accounts.iam.gserviceaccount.com <golang-scoped@luci-project-accounts.iam.gserviceaccount.com>

cmd/worker/main.go

diff --git a/cmd/worker/main.go b/cmd/worker/main.go
index e6644ce..f1d5409 100644
--- a/cmd/worker/main.go
+++ b/cmd/worker/main.go
@@ -50,9 +50,8 @@
 	// Ordinarily, index polling is initiated by a separate scheduler that calls
 	// /poll. But for convenience, you can instead have the worker periodically
 	// do the same.
-	pollIndexPeriod  = flag.Duration("poll_index_period", 0, "when set >0, schedules an index poll at this period")
-	pollIndexLimit   = flag.Int("poll_index_limit", 10, "the amount of modules to fetch from the index when periodically polling")
-	pollIndexHorizon = flag.Duration("poll_index_horizon", time.Hour, "the amount of time ago to request modules each iteration when periodically polling")
+	pollIndexPeriod = flag.Duration("poll_index_period", 0, "when set >0, schedules an index poll at this period")
+	pollIndexLimit  = flag.Int("poll_index_limit", 10, "the amount of modules to fetch from the index when periodically polling")
 
 	// Ordinarily, module version process enqueueing is initiated by a separate
 	// scheduler that calls /enqueue. But for convenience, you can instead have
@@ -175,7 +174,7 @@
 
 	if *pollIndexPeriod != 0 {
 		go func() {
-			log.Infof(ctx, "starting periodic index polling. period=%v, limit=%v, horizon=%v", *pollIndexPeriod, *pollIndexLimit, *pollIndexHorizon)
+			log.Infof(ctx, "starting periodic index polling. period=%v, limit=%v", *pollIndexPeriod, *pollIndexLimit)
 			ticker := time.NewTicker(*pollIndexPeriod)
 			for {
 				select {
@@ -183,7 +182,11 @@
 					log.Warningf(ctx, "cancelling periodic index polling: %v", ctx.Err())
 					return
 				case <-ticker.C:
-					since := time.Now().Add(-1 * *pollIndexHorizon)
+					since, err := db.LatestIndexTimestamp(ctx)
+					if err != nil {
+						log.Warningf(ctx, "error getting latest index timestamp: %v", err)
+						continue
+					}
 					if err := server.PollIndex(ctx, since, *pollIndexLimit); err != nil {
 						log.Warningf(ctx, "error during periodic index polling: %v", err)
 					}