29813735250

Committed 21 Jul 2026 08:17AM UTC coverage: 62.421%. First build

Build # 29813735250

Build Type

Pull #583

github

Committed by

web-flow

Commit Message

chore(lint): fix all golangci-lint issues for v2.12 (#584)

Pull Request Pull Request #583: feat: stage upgrade

Coverage Stats

1584 of 2681 new or added lines in 47 files covered. (59.08%)

11795 of 18896 relevant lines covered (62.42%)

45.45 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

72.73

/api/jobqueue.go

package api

import (
        "fmt"
        "sync"

        "github.com/ethereum/go-ethereum/common"
        "github.com/vocdoni/saas-backend/db"
        "go.vocdoni.io/dvote/log"
)

// orgTxMutex hands out a per-organization mutex so the build->sign->submit pipeline for
// backend-submitted txs (publish and status change) is serialized per org. Two concurrent
// such requests for the same org would otherwise read the same account nonce and sign
// conflicting transactions. In-process only — a multi-instance deployment would
// need a distributed lock, matching the single-instance assumption of db.keysLock. The
// locks map grows unbounded; an org count high enough to matter is not realistic here.
type orgTxMutex struct {
        mu    sync.Mutex
        locks map[common.Address]*sync.Mutex
}

func newOrgTxMutex() *orgTxMutex {
        return &orgTxMutex{locks: make(map[common.Address]*sync.Mutex)}
}

// lock acquires and returns the mutex for addr. Because submit runs on a worker, the
// caller hands the returned mutex to the worker, which Unlocks it after the on-chain
// submit completes — the lock is therefore held across the async hand-off.
func (o *orgTxMutex) lock(addr common.Address) *sync.Mutex {
        o.mu.Lock()
        m, ok := o.locks[addr]
        if !ok {
                m = &sync.Mutex{}
                o.locks[addr] = m
        }
        o.mu.Unlock()
        m.Lock()
        return m
}

// pool sizes are consts; promote to config only if tuning is needed.
const (
        // txQueueSize bounds the number of queued-but-not-yet-running tx tasks.
        txQueueSize = 100
        // txQueueWorkers caps concurrent on-chain submits so a chain stall cannot drain
        // the router's shared request budget or starve the public CSP voter path.
        txQueueWorkers = 16
)

// txTask is a unit of background transaction work. run performs the on-chain submit
// plus any post-submit DB writes, returning the job result on success or an error on
// failure. The worker records the terminal outcome via db.SetJobStatus.
type txTask struct {
        jobID string
        run   func() (*db.JobResult, error)
}

// startTxQueue creates the buffered queue and launches the worker pool. Called once
// from New(). No graceful drain — on process exit in-flight tasks die and
// their jobs stay `pending`; add a Stop()/drain only if that ceiling starts to bite.
func (a *API) startTxQueue() {
        a.txQueue = make(chan txTask, txQueueSize)
        for range txQueueWorkers {
                go a.txWorker()
        }
}

// txWorker runs queued tasks and records each outcome on the job row.
func (a *API) txWorker() {
        for task := range a.txQueue {
                a.runTxTask(task)
        }
}

// runTxTask runs one task and records its outcome, recovering from a panic so a single bad task
// marks its job failed instead of crashing the whole worker pool (and process).
func (a *API) runTxTask(task txTask) {
        defer func() {
                if r := recover(); r != nil {
                        log.Errorw(fmt.Errorf("tx task %s panicked: %v", task.jobID, r), "tx task panicked")
                        if e := a.db.SetJobStatus(task.jobID, db.JobStatusFailed, nil, fmt.Sprintf("panic: %v", r)); e != nil {
                                log.Warnw("could not record panicked job", "jobId", task.jobID, "error", e)
                        }
                }
        }()
        result, err := task.run()
        if err != nil {
                if e := a.db.SetJobStatus(task.jobID, db.JobStatusFailed, nil, err.Error()); e != nil {
                        log.Warnw("could not record failed job", "jobId", task.jobID, "error", e)
                }
                return
        }
        if e := a.db.SetJobStatus(task.jobID, db.JobStatusCompleted, result, ""); e != nil {
                log.Warnw("could not record completed job", "jobId", task.jobID, "error", e)
        }
}

// enqueueTx hands a task to the worker pool without blocking. It returns false when
// the queue is full so the caller can respond 503.
func (a *API) enqueueTx(task txTask) bool {
        select {
        case a.txQueue <- task:
                return true
        default:
                return false
        }
}

1	package api
2
3	import (
4	"fmt"
5	"sync"
6
7	"github.com/ethereum/go-ethereum/common"
8	"github.com/vocdoni/saas-backend/db"
9	"go.vocdoni.io/dvote/log"
10	)
11
12	// orgTxMutex hands out a per-organization mutex so the build->sign->submit pipeline for
13	// backend-submitted txs (publish and status change) is serialized per org. Two concurrent
14	// such requests for the same org would otherwise read the same account nonce and sign
15	// conflicting transactions. In-process only — a multi-instance deployment would
16	// need a distributed lock, matching the single-instance assumption of db.keysLock. The
17	// locks map grows unbounded; an org count high enough to matter is not realistic here.
18	type orgTxMutex struct {
19	mu sync.Mutex
20	locks map[common.Address]*sync.Mutex
21	}
22
23	func newOrgTxMutex() *orgTxMutex {	1✔
24	return &orgTxMutex{locks: make(map[common.Address]*sync.Mutex)}	1✔
25	}	1✔
26
27	// lock acquires and returns the mutex for addr. Because submit runs on a worker, the
28	// caller hands the returned mutex to the worker, which Unlocks it after the on-chain
29	// submit completes — the lock is therefore held across the async hand-off.
30	func (o orgTxMutex) lock(addr common.Address) sync.Mutex {	18✔
31	o.mu.Lock()	18✔
32	m, ok := o.locks[addr]	18✔
33	if !ok {	29✔
34	m = &sync.Mutex{}	11✔
35	o.locks[addr] = m	11✔
36	}	11✔
37	o.mu.Unlock()	18✔
38	m.Lock()	18✔
39	return m	18✔
40	}
41
42	// pool sizes are consts; promote to config only if tuning is needed.
43	const (
44	// txQueueSize bounds the number of queued-but-not-yet-running tx tasks.
45	txQueueSize = 100
46	// txQueueWorkers caps concurrent on-chain submits so a chain stall cannot drain
47	// the router's shared request budget or starve the public CSP voter path.
48	txQueueWorkers = 16
49	)
50
51	// txTask is a unit of background transaction work. run performs the on-chain submit
52	// plus any post-submit DB writes, returning the job result on success or an error on
53	// failure. The worker records the terminal outcome via db.SetJobStatus.
54	type txTask struct {
55	jobID string
56	run func() (*db.JobResult, error)
57	}
58
59	// startTxQueue creates the buffered queue and launches the worker pool. Called once
60	// from New(). No graceful drain — on process exit in-flight tasks die and
61	// their jobs stay `pending`; add a Stop()/drain only if that ceiling starts to bite.
62	func (a *API) startTxQueue() {	1✔
63	a.txQueue = make(chan txTask, txQueueSize)	1✔
64	for range txQueueWorkers {	17✔
65	go a.txWorker()	16✔
66	}	16✔
67	}
68
69	// txWorker runs queued tasks and records each outcome on the job row.
70	func (a *API) txWorker() {	16✔
71	for task := range a.txQueue {	38✔
72	a.runTxTask(task)	22✔
73	}	22✔
74	}
75
76	// runTxTask runs one task and records its outcome, recovering from a panic so a single bad task
77	// marks its job failed instead of crashing the whole worker pool (and process).
78	func (a *API) runTxTask(task txTask) {	22✔
79	defer func() {	44✔
80	if r := recover(); r != nil {	22✔
NEW 81	log.Errorw(fmt.Errorf("tx task %s panicked: %v", task.jobID, r), "tx task panicked")	×
NEW 82	if e := a.db.SetJobStatus(task.jobID, db.JobStatusFailed, nil, fmt.Sprintf("panic: %v", r)); e != nil {	×
NEW 83	log.Warnw("could not record panicked job", "jobId", task.jobID, "error", e)	×
84	}	×
85	}
86	}()
87	result, err := task.run()	22✔
88	if err != nil {	22✔
NEW 89	if e := a.db.SetJobStatus(task.jobID, db.JobStatusFailed, nil, err.Error()); e != nil {	×
NEW 90	log.Warnw("could not record failed job", "jobId", task.jobID, "error", e)	×
91	}	×
NEW 92	return	×
93	}
94	if e := a.db.SetJobStatus(task.jobID, db.JobStatusCompleted, result, ""); e != nil {	22✔
NEW 95	log.Warnw("could not record completed job", "jobId", task.jobID, "error", e)	×
96	}	×
97	}
98
99	// enqueueTx hands a task to the worker pool without blocking. It returns false when
100	// the queue is full so the caller can respond 503.
101	func (a *API) enqueueTx(task txTask) bool {	22✔
102	select {	22✔
103	case a.txQueue <- task:	22✔
104	return true	22✔
105	default:	×
106	return false	×
107	}
108	}

vocdoni / saas-backend / 29813735250

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous