Drop a node's leftover usage of a finished period instead of adding it to the new one

With several nodes, counts one node admitted in the last seconds of a key's period could be written after another node moved the row to the new period, and were added to the new day. They are the old day's usage, so they are dropped: an undercount of at most one sync interval, never a double count.

limit/service_test.go

@@ -902,8 +902,8 @@ func TestService_Persist_LeftoverAfterRollover(t *testing.T) {
 		}, usageRowOf(t, store, "a"))
 		require.Equal(t, int64(2), s.Usage("a").Count(KindMessages))
 
-		// Another node rolled first: a leftover of the old period lands in the new one, and the
-		// row keeps the new period
+		// Another node rolled first: a leftover of the old period is dropped, and the row keeps the
+		// new period and its counts
 		start = periodStart("b", time.Unix(dbNow, 0))
 		require.True(t, s.Allow("b", rule, 3).OK)
 		_, err = store.write(context.Background(), []usageRow{
@@ -918,7 +918,7 @@ func TestService_Persist_LeftoverAfterRollover(t *testing.T) {
 		require.Equal(t, usageRow{
 			key:         "b",
 			periodStart: start + daySeconds,
-			counts:      [numDaily]int64{7, 0, 0},
+			counts:      [numDaily]int64{4, 0, 0},
 		}, usageRowOf(t, store, "b"))
 	})
 }

limit/store.go

@@ -49,8 +49,9 @@ type queries struct {
 }
 
 // write adds the rows' counts to their keys' rows if the periods match, or replaces them if the
-// row's period is older. Rows of an older period than the stored one are added to the stored
-// one (a node's leftover of a finished period, see the usage queries). Rows are written sorted
+// row's period is older. Rows of an older period than the stored one are dropped: a node's
+// leftover of a finished period is the old day's usage, and adding it to the new day would count
+// it twice (see the usage queries). Rows are written sorted
 // by key in chunks, so two nodes lock overlapping keys in the same order; it returns how many
 // rows (in sorted order) are committed: on an error, the chunks before it are, plus the failed
 // chunk if it may have been written anyway (errMaybeWritten). The rows' updated_at is now (unix

limit/store_postgres.go

@@ -11,7 +11,7 @@ import (
 )
 
 // The upsert is one fixed statement per chunk (unnest), because SQL built per call bloats pgx's statement cache.
-// Same period: add; a newer one: replace; an older one (a leftover): add to the stored period.
+// Same period: add; a newer one: replace; an older one (a node's leftover of the old day): drop it.
 const (
 	postgresSelectSinceQuery = `SELECT key, period_start, messages, emails, calls FROM usage WHERE updated_at >= $1`
 	postgresUpsertQuery      = `
@@ -19,9 +19,9 @@ const (
 		SELECT key, ps, m, e, c, $6::BIGINT
 		FROM unnest($1::TEXT[], $2::BIGINT[], $3::BIGINT[], $4::BIGINT[], $5::BIGINT[]) AS d(key, ps, m, e, c)
 		ON CONFLICT (key) DO UPDATE SET
-			messages = CASE WHEN u.period_start >= EXCLUDED.period_start THEN u.messages + EXCLUDED.messages ELSE EXCLUDED.messages END,
-			emails = CASE WHEN u.period_start >= EXCLUDED.period_start THEN u.emails + EXCLUDED.emails ELSE EXCLUDED.emails END,
-			calls = CASE WHEN u.period_start >= EXCLUDED.period_start THEN u.calls + EXCLUDED.calls ELSE EXCLUDED.calls END,
+			messages = CASE WHEN u.period_start = EXCLUDED.period_start THEN u.messages + EXCLUDED.messages WHEN u.period_start < EXCLUDED.period_start THEN EXCLUDED.messages ELSE u.messages END,
+			emails = CASE WHEN u.period_start = EXCLUDED.period_start THEN u.emails + EXCLUDED.emails WHEN u.period_start < EXCLUDED.period_start THEN EXCLUDED.emails ELSE u.emails END,
+			calls = CASE WHEN u.period_start = EXCLUDED.period_start THEN u.calls + EXCLUDED.calls WHEN u.period_start < EXCLUDED.period_start THEN EXCLUDED.calls ELSE u.calls END,
 			period_start = GREATEST(u.period_start, EXCLUDED.period_start),
 			updated_at = EXCLUDED.updated_at
 	`

limit/store_sqlite.go

@@ -15,9 +15,9 @@ const (
 		INSERT INTO usage (key, period_start, messages, emails, calls, updated_at)
 		VALUES (?, ?, ?, ?, ?, ?)
 		ON CONFLICT (key) DO UPDATE SET
-			messages = CASE WHEN period_start >= excluded.period_start THEN messages + excluded.messages ELSE excluded.messages END,
-			emails = CASE WHEN period_start >= excluded.period_start THEN emails + excluded.emails ELSE excluded.emails END,
-			calls = CASE WHEN period_start >= excluded.period_start THEN calls + excluded.calls ELSE excluded.calls END,
+			messages = CASE WHEN period_start = excluded.period_start THEN messages + excluded.messages WHEN period_start < excluded.period_start THEN excluded.messages ELSE messages END,
+			emails = CASE WHEN period_start = excluded.period_start THEN emails + excluded.emails WHEN period_start < excluded.period_start THEN excluded.emails ELSE emails END,
+			calls = CASE WHEN period_start = excluded.period_start THEN calls + excluded.calls WHEN period_start < excluded.period_start THEN excluded.calls ELSE calls END,
 			period_start = MAX(period_start, excluded.period_start),
 			updated_at = excluded.updated_at
 	`

limit/store_test.go

@@ -100,7 +100,7 @@ func TestStore_WriteAddsSamePeriodReplacesNewer(t *testing.T) {
 			counts:      [numDaily]int64{4, 0, 0},
 		}, rows["user:u_1"])
 
-		// A leftover of an older period lands in the stored, newer one instead of taking the row back
+		// A leftover of an older period is dropped: it is the old day's usage, not the new one's
 		_, err = s.write(ctx, []usageRow{
 			{
 				key:         "user:u_1",
@@ -112,7 +112,7 @@ func TestStore_WriteAddsSamePeriodReplacesNewer(t *testing.T) {
 		require.Equal(t, usageRow{
 			key:         "user:u_1",
 			periodStart: 1000 + daySeconds,
-			counts:      [numDaily]int64{6, 0, 0},
+			counts:      [numDaily]int64{4, 0, 0},
 		}, readAll(t, s, 0)["user:u_1"])
 	})
 }