From 188d0ed895a1a66aba85fcdbae81b7f567f7fbb0 Mon Sep 17 00:00:00 2001 From: bwees Date: Sat, 14 Nov 2026 14:32:09 -0600 Subject: [PATCH] fix(sync): stop resending tombstones that predate a full sync --- apps/gateway/repositories/sync.go | 24 ++++++++++++++++++----- apps/gateway/services/sync.go | 14 ++++++------- docs/design/sync-stream.md | 2 +- test/medium/services/sync_service_test.go | 20 ++++++++++++++----- 4 files changed, 42 insertions(+), 18 deletions(-) diff --git a/apps/gateway/repositories/sync.go b/apps/gateway/repositories/sync.go index 208fc03..186bc9e 100644 --- a/apps/gateway/repositories/sync.go +++ b/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 diff --git a/apps/gateway/services/sync.go b/apps/gateway/services/sync.go index 22cac4c..baa0961 100644 --- a/apps/gateway/services/sync.go +++ b/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 { diff --git a/docs/design/sync-stream.md b/docs/design/sync-stream.md index cfaa401..2c3f130 100644 --- a/docs/design/sync-stream.md +++ b/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. diff --git a/test/medium/services/sync_service_test.go b/test/medium/services/sync_service_test.go index d0a14a7..0a40035 100644 --- a/test/medium/services/sync_service_test.go +++ b/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)) }) })