fix(sync): stop resending tombstones that predate a full sync

apps/gateway/repositories/sync.go

@@ -60,18 +60,32 @@ func Deletes[A sync.Row](ctx context.Context, r *SyncRepository, after, before u
 	return rowsBetween[A](ctx, r.db, sync.DeleteColumn, after, before)
 }
 
+// NewestDelete yields the newest tombstone of A written before `before`, if there is one.
+func NewestDelete[A sync.Row](ctx context.Context, r *SyncRepository, before uuid.UUID) iter.Seq2[A, error] {
+	return scanRows[A](ctx, r.db, func(q *bun.SelectQuery) *bun.SelectQuery {
+		return q.
+			Where("? < ?", bun.Ident(sync.DeleteColumn), before).
+			OrderExpr("? DESC", bun.Ident(sync.DeleteColumn)).
+			Limit(1)
+	})
+}
+
 func rowsBetween[M sync.Row](ctx context.Context, db *bun.DB, column string, after, before uuid.UUID) iter.Seq2[M, error] {
-	return func(yield func(M, error) bool) {
-		var zero M
-		q := db.NewSelect().
-			Model(&zero).
+	return scanRows[M](ctx, db, func(q *bun.SelectQuery) *bun.SelectQuery {
+		q = q.
 			Where("? < ?", bun.Ident(column), before).
 			OrderExpr("? ASC", bun.Ident(column))
 		if after != uuid.Nil {
 			q = q.Where("? > ?", bun.Ident(column), after)
 		}
+		return q
+	})
+}
 
-		rows, err := q.Rows(ctx)
+func scanRows[M sync.Row](ctx context.Context, db *bun.DB, query func(*bun.SelectQuery) *bun.SelectQuery) iter.Seq2[M, error] {
+	return func(yield func(M, error) bool) {
+		var zero M
+		rows, err := query(db.NewSelect().Model(&zero)).Rows(ctx)
 		if err != nil {
 			yield(zero, fmt.Errorf("query %T: %w", zero, err))
 			return

apps/gateway/services/sync.go

@@ -72,7 +72,7 @@ func NewSyncStream[M, A sync.Row, U, D any](
 		Deletes: deletes,
 		Model:   reflect.TypeFor[M](),
 		run: func(ctx context.Context, repo *repositories.SyncRepository, cursor syncCursor, nowID uuid.UUID, send SyncSender) error {
-			for row, err := range deletesAfter[A](ctx, repo, newestID(cursor[upserts], cursor[deletes]), nowID) {
+			for row, err := range deletesFor[A](ctx, repo, cursor[upserts], cursor[deletes], nowID) {
 				if err != nil {
 					return err
 				}
@@ -93,13 +93,13 @@ func NewSyncStream[M, A sync.Row, U, D any](
 	}
 }
 
-// A client with no ack for a stream holds none of its rows, so it needs no tombstones.
-// Tombstones older than its upsert ack were already left out of the upserts it got.
-func deletesAfter[A sync.Row](ctx context.Context, repo *repositories.SyncRepository, after, nowID uuid.UUID) iter.Seq2[A, error] {
-	if after == uuid.Nil {
-		return func(func(A, error) bool) {}
+// A client with no upsert ack holds none of a stream's rows, so it gets only the newest
+// tombstone. Its ack tells the next sync that every older tombstone predates the rows it got.
+func deletesFor[A sync.Row](ctx context.Context, repo *repositories.SyncRepository, upsertAck, deleteAck, nowID uuid.UUID) iter.Seq2[A, error] {
+	if upsertAck == uuid.Nil {
+		return repositories.NewestDelete[A](ctx, repo, nowID)
 	}
-	return repositories.Deletes[A](ctx, repo, after, nowID)
+	return repositories.Deletes[A](ctx, repo, newestID(upsertAck, deleteAck), nowID)
 }
 
 func newestID(a, b uuid.UUID) uuid.UUID {

docs/design/sync-stream.md

@@ -103,7 +103,7 @@ The server parses the acks into a cursor keyed by entity type. It never trusts a
    2. upserts: rows with `updateId > ack`, `updateId < nowId`, ordered by `updateId`
 5. Send `SyncCompleteV1` with `nowId` and end the stream.
 
-With no ack for an entity type, its query starts from the beginning. A client with no acks at all gets a full sync. It gets no deletes that matter, because it has nothing to delete.
+With no ack for an entity type, its query starts from the beginning. A client with no acks at all gets a full sync. With no upsert ack for a type, the client has nothing to delete, so it gets only the newest tombstone. That line's ack keeps the next sync from sending tombstones written before this one.
 
 The fixed order follows foreign keys, so parents arrive before children: routes, directions, stops, alerts, alert directions, stop schedules. All versions of a resource sit at the same position. The order of `types` in the request does not matter.
 

test/medium/services/sync_service_test.go

@@ -275,10 +275,11 @@ var _ = Describe("SyncService", Label("medium", "service"), func() {
 			kept = test.CreateNetwork(bundb, "02")
 		})
 
-		It("are not sent to a client with no acks", func() {
+		It("are sent to a client with no acks only as the newest, which starts its delete ack", func() {
 			lines := client.sync(requests.RoutesV1)
 
-			Expect(types(lines)).To(Equal([]dtos.SyncEntityType{entities.RouteV1, entities.SyncCompleteV1}))
+			Expect(types(lines)).To(Equal([]dtos.SyncEntityType{entities.RouteDeleteV1, entities.RouteV1, entities.SyncCompleteV1}))
+			Expect(lines[0].Data).To(Equal(dtos.SyncRouteDeleteV1{RouteID: deleted.Route.ID}))
 		})
 
 		It("older than the upsert ack are not sent to a client with no delete ack", func() {
@@ -290,6 +291,14 @@ var _ = Describe("SyncService", Label("medium", "service"), func() {
 			Expect(types(lines)).To(Equal([]dtos.SyncEntityType{entities.RouteDeleteV1, entities.SyncCompleteV1}))
 			Expect(lines[0].Data).To(Equal(dtos.SyncRouteDeleteV1{RouteID: kept.Route.ID}))
 		})
+
+		It("newer than every upserted row are not sent again after a full sync", func() {
+			pruned := test.CreateNetwork(bundb, "03")
+			test.Exec(bundb, `DELETE FROM "route" WHERE "id" = ?`, pruned.Route.ID)
+			client.sync(allTypes...)
+
+			Expect(types(client.sync(allTypes...))).To(Equal([]dtos.SyncEntityType{entities.SyncCompleteV1}))
+		})
 	})
 
 	Describe("acks", func() {
@@ -471,7 +480,7 @@ gateway_syncs_total{kind="reset"} 1
 			Expect(lines[0].Data.(dtos.SyncRouteV2).ID).To(Equal(network.Route.ID))
 		})
 
-		It("ignores V1 delete acks and sends no tombstones", func() {
+		It("ignores V1 delete acks and sends only the newest tombstone", func() {
 			deleted := test.CreateNetwork(bundb, "01")
 			kept := test.CreateNetwork(bundb, "02")
 			client.sync(requests.RoutesV1)
@@ -481,8 +490,9 @@ gateway_syncs_total{kind="reset"} 1
 
 			lines := client.stream(dtos.SyncRequest{Protocol: dtos.SyncProtocolVersions.V1, Types: []dtos.SyncRequestType{requests.RoutesV2}, Acks: client.storedAcks()})
 
-			Expect(types(lines)).To(Equal([]dtos.SyncEntityType{entities.RouteV2, entities.SyncCompleteV1}))
-			Expect(lines[0].Data.(dtos.SyncRouteV2).ID).To(Equal(kept.Route.ID))
+			Expect(types(lines)).To(Equal([]dtos.SyncEntityType{entities.RouteDeleteV2, entities.RouteV2, entities.SyncCompleteV1}))
+			Expect(lines[0].Data).To(Equal(dtos.SyncRouteDeleteV1{RouteID: deleted.Route.ID}))
+			Expect(lines[1].Data.(dtos.SyncRouteV2).ID).To(Equal(kept.Route.ID))
 		})
 	})