rouggy/OpsLog
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 30 Wiki Activity

fix: serialise auto-uploads so a burst doesn't 403 and leave QSOs at R

Browse Source 
A QSO logged via UDP / ADIF-import fired `go m.upload` per QSO, so a pileup or
an ADIF-import burst hit a service with dozens of concurrent requests at once.
Club Log's nginx answers that burst with 403; the upload counts as failed and
the QSO stays at "R" while a lucky few go through — the real cause behind the
"it's uploaded but shows R" report (not the earlier display race).

Route all immediate/delayed auto-uploads through a single serialised worker
(one at a time, 250ms apart) so no burst ever trips a per-IP rate limiter.
upload() now reports whether a failure is transient (HTTP/rate-limit) vs
permanent (not eligible, wrong station callsign, no record); transient ones
are retried with exponential back-off (up to 4 attempts, 1s/2s/4s), so a QSO
that hit a momentary 403 ends up marked instead of stuck at R. A full queue
falls back to a goroutine rather than dropping an upload.
This commit is contained in:
rouggy
2026-07-19 03:05:02 +02:00
parent d38c783dcc
commit bd9e091e65
1 changed files with 85 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/extsvc/manager.go
+85 -20
View File
@@ -87,17 +87,76 @@ type Manager struct {
mu sync.Mutex
cfg ExternalServices
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.
type uploadJob struct {
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 {
if deps.Client == nil {
deps.Client = &http.Client{Timeout: 20 * time.Second}
}
return &Manager{
m := &Manager{
deps: deps,
// Seeded from the clock; the delay only needs to be unpredictable
// 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.
func (m *Manager) enqueueUpload(svc Service, id int64, cfg ServiceConfig) {
m.enqueueJob(uploadJob{svc: svc, id: id, cfg: cfg})
}
// enqueueJob queues a job (possibly a retry). If the queue is somehow full (an
// enormous burst), it falls back to a goroutine rather than dropping the upload —
// a dropped upload would leave the QSO stuck at "R".
func (m *Manager) enqueueJob(job uploadJob) {
select {
case m.uploadCh <- job:
default:
go m.upload(job.svc, job.id, job.cfg)
}
}
@@ -175,15 +234,17 @@ func (m *Manager) route(svc Service, id int64, cfg ServiceConfig) {
m.scheduleUpload(svc, id, cfg)
}
// scheduleUpload either uploads now (immediate) or arms a timer (delayed).
// scheduleUpload either queues the upload now (immediate) or arms a timer that
// queues it later (delayed). Both go through the serialised worker so uploads are
// never fired concurrently in a burst.
func (m *Manager) scheduleUpload(svc Service, id int64, cfg ServiceConfig) {
if cfg.UploadMode == ModeDelayed {
d := m.delaySeconds()
m.logf("extsvc: %s upload of QSO %d scheduled in %s", svc, id, d)
time.AfterFunc(d, func() { m.upload(svc, id, cfg) })
time.AfterFunc(d, func() { m.enqueueUpload(svc, id, cfg) })
return
}
go m.upload(svc, id, cfg)
m.enqueueUpload(svc, id, cfg)
}
// onCloseServices returns the services configured for on-close auto-upload,
@@ -243,25 +304,25 @@ func (m *Manager) FlushOnClose() int {
uploaded += m.flushLoTWBatch(ids, cfg.LoTW)
case ServiceQRZ:
for _, id := range ids {
if m.upload(svc, id, cfg.QRZ) {
if ok, _ := m.upload(svc, id, cfg.QRZ); ok {
uploaded++
}
}
case ServiceClublog:
for _, id := range ids {
if m.upload(svc, id, cfg.Clublog) {
if ok, _ := m.upload(svc, id, cfg.Clublog); ok {
uploaded++
}
}
case ServiceHRDLog:
for _, id := range ids {
if m.upload(svc, id, cfg.HRDLog) {
if ok, _ := m.upload(svc, id, cfg.HRDLog); ok {
uploaded++
}
}
case ServiceEQSL:
for _, id := range ids {
if m.upload(svc, id, cfg.EQSL) {
if ok, _ := m.upload(svc, id, cfg.EQSL); ok {
uploaded++
}
}
@@ -312,12 +373,16 @@ func (m *Manager) flushLoTWBatch(ids []int64, cfg ServiceConfig) int {
// upload performs the actual push and returns true on success. It builds a
// fresh, lifecycle-independent context so a delayed upload still completes
// even if it fires close to shutdown.
func (m *Manager) upload(svc Service, id int64, cfg ServiceConfig) bool {
// upload performs one upload. It returns ok=true when the QSO was uploaded (and
// marked), and retryable=true when it failed in a way worth retrying later (an
// HTTP/service error such as a rate-limit 403) as opposed to a permanent skip
// (not eligible, wrong station callsign, no record).
func (m *Manager) upload(svc Service, id int64, cfg ServiceConfig) (ok bool, retryable bool) {
// Skip QSOs that aren't eligible (already sent, or sent status doesn't
// match the configured Upload flag).
if m.deps.ShouldUpload != nil && !m.deps.ShouldUpload(svc, id) {
m.logf("extsvc: %s upload of QSO %d skipped (not eligible)", svc, id)
return false
return false, false
}
// Station-callsign guard. Each logbook belongs to one callsign:
@@ -345,7 +410,7 @@ func (m *Manager) upload(svc Service, id int64, cfg ServiceConfig) bool {
if m.deps.NotifyError != nil {
m.deps.NotifyError(svc, id, err)
}
return false
return false, false
}
}
@@ -360,7 +425,7 @@ func (m *Manager) upload(svc Service, id int64, cfg ServiceConfig) bool {
record, ok := m.deps.BuildADIF(id, cfg.ForceStationCallsign)
if !ok {
m.logf("extsvc: %s upload of QSO %d skipped (no record)", svc, id)
return false
return false, false
}
res, err = UploadQRZ(ctx, m.deps.Client, cfg.APIKey, record)
case ServiceClublog:
@@ -369,7 +434,7 @@ func (m *Manager) upload(svc Service, id int64, cfg ServiceConfig) bool {
record, ok := m.deps.BuildADIF(id, "")
if !ok {
m.logf("extsvc: %s upload of QSO %d skipped (no record)", svc, id)
return false
return false, false
}
res, err = UploadClublog(ctx, m.deps.Client, cfg, record)
case ServiceLoTW:
@@ -378,7 +443,7 @@ func (m *Manager) upload(svc Service, id int64, cfg ServiceConfig) bool {
record, ok := m.deps.BuildADIF(id, cfg.ForceStationCallsign)
if !ok {
m.logf("extsvc: %s upload of QSO %d skipped (no record)", svc, id)
return false
return false, false
}
res, err = UploadLoTW(ctx, cfg, "", record)
case ServiceHRDLog:
@@ -387,7 +452,7 @@ func (m *Manager) upload(svc Service, id int64, cfg ServiceConfig) bool {
record, ok := m.deps.BuildADIF(id, "")
if !ok {
m.logf("extsvc: %s upload of QSO %d skipped (no record)", svc, id)
return false
return false, false
}
res, err = UploadHRDLog(ctx, m.deps.Client, cfg.Callsign, cfg.Code, record)
case ServiceEQSL:
@@ -396,11 +461,11 @@ func (m *Manager) upload(svc Service, id int64, cfg ServiceConfig) bool {
record, ok := m.deps.BuildADIF(id, "")
if !ok {
m.logf("extsvc: %s upload of QSO %d skipped (no record)", svc, id)
return false
return false, false
}
res, err = UploadEQSL(ctx, m.deps.Client, cfg.Username, cfg.Password, cfg.QTHNickname, record)
default:
return false
return false, false
}
if err != nil || !res.OK {
@@ -411,12 +476,12 @@ func (m *Manager) upload(svc Service, id int64, cfg ServiceConfig) bool {
if m.deps.NotifyError != nil {
m.deps.NotifyError(svc, id, err)
}
return false
return false, true // transient (rate-limit / network) → worth a retry
}
m.logf("extsvc: %s upload of QSO %d OK (logid=%q)", svc, id, res.LogID)
if m.deps.MarkUploaded != nil {
m.deps.MarkUploaded(svc, id, res.LogID)
}
return true
return true, false
}