fix: revert upload serialisation — it stranded QSOs at R (regression)
The single-worker upload queue introduced head-of-line blocking: one slow upload (LoTW signing via TQSL, or a service on a 30 s timeout) stalled the whole queue, so every following QSO's Club Log upload waited and the rows sat at "R". Operators on the older, concurrent build had no such problem — the tell that this was a fresh regression. Back to one goroutine per upload (never serialised), so a slow upload never holds up the rest, while keeping the transient-failure retry with back-off (concurrent, so it can't block anything).
This commit is contained in:
+23
-61
@@ -87,77 +87,39 @@ type Manager struct {
|
|||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
cfg ExternalServices
|
cfg ExternalServices
|
||||||
rnd *rand.Rand
|
rnd *rand.Rand
|
||||||
|
|
||||||
// uploadCh serialises immediate auto-uploads through a single worker. Firing a
|
|
||||||
// goroutine per QSO meant a pileup / ADIF-import burst hit a service with dozens
|
|
||||||
// of concurrent requests at once — Club Log's nginx answers that with 403, the
|
|
||||||
// upload is counted as failed and the QSO stays at "R" despite the others going
|
|
||||||
// through. One-at-a-time with a small gap keeps every upload under the limit.
|
|
||||||
uploadCh chan uploadJob
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// uploadJob is one queued auto-upload.
|
// maxUploadAttempts bounds retries of a transient upload failure.
|
||||||
type uploadJob struct {
|
const maxUploadAttempts = 4
|
||||||
svc Service
|
|
||||||
id int64
|
|
||||||
cfg ServiceConfig
|
|
||||||
attempt int // 0 on first try; incremented on each retry
|
|
||||||
}
|
|
||||||
|
|
||||||
// uploadGap spaces serialized uploads so a burst never trips a service's per-IP
|
|
||||||
// rate limiter. maxUploadAttempts bounds retries of a transient failure.
|
|
||||||
const (
|
|
||||||
uploadGap = 250 * time.Millisecond
|
|
||||||
maxUploadAttempts = 4
|
|
||||||
)
|
|
||||||
|
|
||||||
func NewManager(deps Deps) *Manager {
|
func NewManager(deps Deps) *Manager {
|
||||||
if deps.Client == nil {
|
if deps.Client == nil {
|
||||||
deps.Client = &http.Client{Timeout: 20 * time.Second}
|
deps.Client = &http.Client{Timeout: 20 * time.Second}
|
||||||
}
|
}
|
||||||
m := &Manager{
|
return &Manager{
|
||||||
deps: deps,
|
deps: deps,
|
||||||
// Seeded from the clock; the delay only needs to be unpredictable
|
// Seeded from the clock; the delay only needs to be unpredictable
|
||||||
// enough to spread bursts, not cryptographically random.
|
// enough to spread bursts, not cryptographically random.
|
||||||
rnd: rand.New(rand.NewSource(time.Now().UnixNano())),
|
rnd: rand.New(rand.NewSource(time.Now().UnixNano())),
|
||||||
uploadCh: make(chan uploadJob, 4096),
|
|
||||||
}
|
|
||||||
go m.uploadWorker()
|
|
||||||
return m
|
|
||||||
}
|
|
||||||
|
|
||||||
// uploadWorker drains the queue one upload at a time, spacing them so a burst of
|
|
||||||
// freshly-logged QSOs can't hammer (and get 403'd by) a service. A transient
|
|
||||||
// failure is re-queued with an exponential back-off, so a QSO that hit a
|
|
||||||
// momentary rate-limit still ends up marked instead of stuck at "R".
|
|
||||||
func (m *Manager) uploadWorker() {
|
|
||||||
for job := range m.uploadCh {
|
|
||||||
ok, retryable := m.upload(job.svc, job.id, job.cfg)
|
|
||||||
if !ok && retryable && job.attempt+1 < maxUploadAttempts {
|
|
||||||
next := job
|
|
||||||
next.attempt++
|
|
||||||
backoff := time.Duration(1<<uint(job.attempt)) * time.Second // 1s, 2s, 4s…
|
|
||||||
m.logf("extsvc: %s upload of QSO %d will retry (attempt %d) in %s", job.svc, job.id, next.attempt+1, backoff)
|
|
||||||
time.AfterFunc(backoff, func() { m.enqueueJob(next) })
|
|
||||||
}
|
|
||||||
time.Sleep(uploadGap)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// enqueueUpload queues a first-attempt upload without blocking the logging path.
|
// attemptUpload uploads a QSO in its OWN goroutine and, on a TRANSIENT failure
|
||||||
func (m *Manager) enqueueUpload(svc Service, id int64, cfg ServiceConfig) {
|
// (rate-limit / network), re-arms itself with exponential back-off. Each upload is
|
||||||
m.enqueueJob(uploadJob{svc: svc, id: id, cfg: cfg})
|
// independent — never serialised through a shared worker, because a single slow
|
||||||
}
|
// upload (LoTW signs via TQSL; a service on a 30 s timeout) would otherwise block
|
||||||
|
// every following QSO's upload and strand them all at "R" (the regression that hit
|
||||||
// enqueueJob queues a job (possibly a retry). If the queue is somehow full (an
|
// the operator on the newest build while everyone on the old concurrent path was
|
||||||
// enormous burst), it falls back to a goroutine rather than dropping the upload —
|
// fine).
|
||||||
// a dropped upload would leave the QSO stuck at "R".
|
func (m *Manager) attemptUpload(svc Service, id int64, cfg ServiceConfig, attempt int) {
|
||||||
func (m *Manager) enqueueJob(job uploadJob) {
|
go func() {
|
||||||
select {
|
ok, retryable := m.upload(svc, id, cfg)
|
||||||
case m.uploadCh <- job:
|
if !ok && retryable && attempt+1 < maxUploadAttempts {
|
||||||
default:
|
backoff := time.Duration(1<<uint(attempt)) * time.Second // 1s, 2s, 4s…
|
||||||
go m.upload(job.svc, job.id, job.cfg)
|
m.logf("extsvc: %s upload of QSO %d will retry (attempt %d) in %s", svc, id, attempt+2, backoff)
|
||||||
|
time.AfterFunc(backoff, func() { m.attemptUpload(svc, id, cfg, attempt+1) })
|
||||||
}
|
}
|
||||||
|
}()
|
||||||
}
|
}
|
||||||
|
|
||||||
func (m *Manager) logf(format string, args ...any) {
|
func (m *Manager) logf(format string, args ...any) {
|
||||||
@@ -234,17 +196,17 @@ func (m *Manager) route(svc Service, id int64, cfg ServiceConfig) {
|
|||||||
m.scheduleUpload(svc, id, cfg)
|
m.scheduleUpload(svc, id, cfg)
|
||||||
}
|
}
|
||||||
|
|
||||||
// scheduleUpload either queues the upload now (immediate) or arms a timer that
|
// scheduleUpload uploads now (immediate) or after a random fuse (delayed). Each
|
||||||
// queues it later (delayed). Both go through the serialised worker so uploads are
|
// upload runs in its own goroutine (attemptUpload) — never serialised — so a slow
|
||||||
// never fired concurrently in a burst.
|
// one never holds up the rest.
|
||||||
func (m *Manager) scheduleUpload(svc Service, id int64, cfg ServiceConfig) {
|
func (m *Manager) scheduleUpload(svc Service, id int64, cfg ServiceConfig) {
|
||||||
if cfg.UploadMode == ModeDelayed {
|
if cfg.UploadMode == ModeDelayed {
|
||||||
d := m.delaySeconds()
|
d := m.delaySeconds()
|
||||||
m.logf("extsvc: %s upload of QSO %d scheduled in %s", svc, id, d)
|
m.logf("extsvc: %s upload of QSO %d scheduled in %s", svc, id, d)
|
||||||
time.AfterFunc(d, func() { m.enqueueUpload(svc, id, cfg) })
|
time.AfterFunc(d, func() { m.attemptUpload(svc, id, cfg, 0) })
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
m.enqueueUpload(svc, id, cfg)
|
m.attemptUpload(svc, id, cfg, 0)
|
||||||
}
|
}
|
||||||
|
|
||||||
// onCloseServices returns the services configured for on-close auto-upload,
|
// onCloseServices returns the services configured for on-close auto-upload,
|
||||||
|
|||||||
Reference in New Issue
Block a user