fix9b2-6: the publish queue sends in the order it queued, whatever the clock says between two versions
Builds: design 47 G §588 (newest first)
This commit is contained in:
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user