From f299a1a8b9265da79f7be66d2afac9bfad45fad1 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 01:31:19 +0100 Subject: [PATCH] fix9b2-6: the publish queue sends in the order it queued, whatever the clock says between two versions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G ยง588 (newest first) --- .../internal/versions/republish.go | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/modules/mesh-delivery-builder/internal/versions/republish.go b/modules/mesh-delivery-builder/internal/versions/republish.go index 07dc30cc..e5950b5d 100644 --- a/modules/mesh-delivery-builder/internal/versions/republish.go +++ b/modules/mesh-delivery-builder/internal/versions/republish.go @@ -118,6 +118,22 @@ type Republisher struct { mu sync.Mutex runs map[string]map[string][]string // node -> holder module -> the registry seats it claims busy map[string]bool + // lastQueued is the newest time a version was queued at: a clock that has not moved between two versions (a coarse clock, + // or a fixed one) must not make the queue's order the order of its keys, so each is queued after the last. + lastQueued time.Time +} + +// queuedAt is the time a version is queued at: now, or a nanosecond after the last one queued when now is not after it, so the +// order the versions were queued in (newest reservation first) is the order they are sent in. +func (r *Republisher) queuedAt() time.Time { + r.mu.Lock() + defer r.mu.Unlock() + at := r.now() + if !at.After(r.lastQueued) { + at = r.lastQueued.Add(time.Nanosecond) + } + r.lastQueued = at + return at } func (r *Republisher) now() time.Time { @@ -349,7 +365,7 @@ func (r *Republisher) enqueue(ctx context.Context, v ReservedVersion, holder str } else if found { return false, nil } - raw, _ := json.Marshal(QueuedPublish{V: 1, Holder: holder, Module: v.Module, Version: v.Version, Artifacts: arts, Queued: r.now()}) + raw, _ := json.Marshal(QueuedPublish{V: 1, Holder: holder, Module: v.Module, Version: v.Version, Artifacts: arts, Queued: r.queuedAt()}) used, err := r.Queue.Bytes(ctx) if err != nil { return false, fmt.Errorf("the size of %s cannot be read: %w", PublishQueueBucket, err)