Skip to content

Commit 1df4a9f

Browse files
jonyoderjonyoder
committed
Better failure reporting from database queue
1 parent e97ce48 commit 1df4a9f

File tree

9 files changed

+89
-208
lines changed

9 files changed

+89
-208
lines changed

‎examples/cmd/markdownRenderer/store/queue.go‎

Lines changed: 0 additions & 110 deletions
Original file line numberDiff line numberDiff line change
@@ -224,35 +224,6 @@ func (conn *store) QueueGroupComplete(id int64) (done bool, cancelled bool, err
224224
return count == 0, cancelled, err
225225
}
226226

227-
func (conn *store) IsQueueAddressComplete(address string) (done bool, err error) {
228-
address = strings.TrimSpace(address)
229-
if address == "" {
230-
return false, errors.New("no address provided for IsQueueAddressComplete")
231-
}
232-
233-
// Include the address in the transaction description to avoid races when
234-
// testing with a special locking connection.
235-
var count int64
236-
var workErr error
237-
conn.db.Transaction(func(tx *gorm.DB) error {
238-
err = tx.Model(&Queue{}).Where("address = ?", address).Count(&count).Error
239-
if err != nil {
240-
return err
241-
}
242-
243-
c2 := &store{
244-
db: tx,
245-
}
246-
workErr = c2.QueueAddressedCheck(address)
247-
return nil
248-
})
249-
250-
if workErr != nil {
251-
err = workErr
252-
}
253-
return count == 0, err
254-
}
255-
256227
func (conn *store) IsQueueAddressInProgress(address string) (bool, error) {
257228
address = strings.TrimSpace(address)
258229
if address == "" {
@@ -568,87 +539,6 @@ func (conn *store) QueueDelete(permitId permit.Permit) (err error) {
568539
})
569540
}
570541

571-
func (conn *store) QueueAddressedComplete(address string, failure error) (err error) {
572-
573-
var bytes []byte
574-
if failure != nil {
575-
queueError, isQueueError := failure.(*queue.QueueError)
576-
if isQueueError {
577-
// Record type of queue.QueueError
578-
bytes, err = json.Marshal(queueError)
579-
} else {
580-
// Record generic error
581-
bytes, err = json.Marshal(&queue.QueueError{
582-
Message: failure.Error(),
583-
})
584-
}
585-
}
586-
// Return if there were any marshaling errors
587-
if err != nil {
588-
return err
589-
}
590-
591-
var postgres bool
592-
593-
err = conn.db.Transaction(func(tx *gorm.DB) error {
594-
if postgres {
595-
// Lock to prevent race. Don't allow reading until this operation is complete.
596-
err = tx.Exec("LOCK queue_failure IN ACCESS EXCLUSIVE MODE").Error
597-
if err != nil {
598-
return err
599-
}
600-
}
601-
602-
// First, delete any references to this address
603-
err = tx.Delete(&QueueFailure{}, "address = ?", address).Error
604-
if err != nil {
605-
return err
606-
}
607-
608-
// Next, if there is an error to report, insert it
609-
if failure != nil {
610-
failure := QueueFailure{
611-
Address: address,
612-
Error: string(bytes),
613-
}
614-
err = tx.Create(&failure).Error
615-
if err != nil {
616-
return err
617-
}
618-
}
619-
return nil
620-
})
621-
return
622-
}
623-
624-
type QueueAddressFailure struct {
625-
Message string `json:"error"`
626-
}
627-
628-
func (err *QueueAddressFailure) Error() string {
629-
return err.Message
630-
}
631-
632-
func (conn *store) QueueAddressedCheck(address string) error {
633-
var failure QueueFailure
634-
err := conn.db.First(&failure, "address = ?", address).Error
635-
if err == gorm.ErrRecordNotFound {
636-
return nil
637-
} else if err != nil {
638-
return err
639-
} else {
640-
if failure.Error != "" {
641-
var queueError queue.QueueError
642-
if err := json.Unmarshal([]byte(failure.Error), &queueError); err != nil {
643-
return fmt.Errorf("error unmarshalling queue.QueueError: %s", err)
644-
} else {
645-
return &queueError
646-
}
647-
}
648-
}
649-
return nil
650-
}
651-
652542
func (conn *store) QueuePermits(name string) ([]dbqueuetypes.QueuePermit, error) {
653543
permits := make([]QueuePermit, 0)
654544

‎graph.svg‎

Lines changed: 7 additions & 7 deletions
Loading

‎pkg/rsqueue/impls/database/dbqueuetypes/types.go‎

Lines changed: 0 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -71,24 +71,13 @@ type QueueStore interface {
7171
// * error - errors
7272
QueuePeek(types ...uint64) (results []queue.QueueWork, err error)
7373

74-
// IsQueueAddressComplete checks to see if an address is done/gone
75-
// Expects:
76-
// * id - the queue item address
77-
// Returns:
78-
// * bool - is the item done/gone?
79-
// * error - errors
80-
IsQueueAddressComplete(address string) (bool, error)
81-
8274
// IsQueueAddressInProgress checks to see if an address is still in progress
8375
// Expects:
8476
// * id - the queue item address
8577
// Returns:
8678
// * bool - is the item still in progress?
8779
// * error - errors
8880
IsQueueAddressInProgress(address string) (bool, error)
89-
90-
// QueueAddressedComplete saves or clears failure information for an addressed item
91-
QueueAddressedComplete(address string, failure error) error
9281
}
9382

9483
type QueueGroupStore interface {

‎pkg/rsqueue/impls/database/go.mod‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,9 +5,9 @@ go 1.20
55
require (
66
github.com/fortytw2/leaktest v1.3.0
77
github.com/google/uuid v1.1.2
8-
github.com/rstudio/platform-lib/pkg/rsnotify v1.4.0
8+
github.com/rstudio/platform-lib/pkg/rsnotify v1.5.1
99
github.com/rstudio/platform-lib/pkg/rsnotify/listeners/local v1.4.1
10-
github.com/rstudio/platform-lib/pkg/rsqueue v0.1.1
10+
github.com/rstudio/platform-lib/pkg/rsqueue v0.3.0
1111
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c
1212
)
1313

‎pkg/rsqueue/impls/database/go.sum‎

Lines changed: 4 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -9,13 +9,11 @@ github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
99
github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
1010
github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
1111
github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
12-
github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
13-
github.com/rstudio/platform-lib/pkg/rsnotify v1.4.0 h1:tMeO7PckDZzDROBmu5nZJ1pORN3dNDWMWv57X0xpMf4=
14-
github.com/rstudio/platform-lib/pkg/rsnotify v1.4.0/go.mod h1:OTMZNgESF0Y2THisqq2gd4P/yF9YLCUW1GNXy5ainYM=
12+
github.com/rstudio/platform-lib/pkg/rsnotify v1.5.1 h1:hkLLKye0iBqW5r5ae6qnX5viU9qj0Eiwh+vUpkuy2zI=
13+
github.com/rstudio/platform-lib/pkg/rsnotify v1.5.1/go.mod h1:OTMZNgESF0Y2THisqq2gd4P/yF9YLCUW1GNXy5ainYM=
1514
github.com/rstudio/platform-lib/pkg/rsnotify/listeners/local v1.4.1 h1:65BsZFFHKW9Bm0MBi3Jw7a72C2XqRnmA+519Cm+6zFk=
1615
github.com/rstudio/platform-lib/pkg/rsnotify/listeners/local v1.4.1/go.mod h1:3fj43uPHSrY30ZJmAgbSmmy9d7heQVnL/MXVwOz+5Pc=
17-
github.com/rstudio/platform-lib/pkg/rsqueue v0.1.1 h1:y6hxq3z1F98KFEnfSSYDm5BP4C9MlSxwnNPKHGluV7w=
18-
github.com/rstudio/platform-lib/pkg/rsqueue v0.1.1/go.mod h1:v47Q+KI/vBl6NLgrMLry6x3J651uhppVvCCGffq/bDI=
19-
gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
16+
github.com/rstudio/platform-lib/pkg/rsqueue v0.3.0 h1:2/risfO4VatW7FcmWeup0Mbx1QlPLYYUZrN8sb8Pt+g=
17+
github.com/rstudio/platform-lib/pkg/rsqueue v0.3.0/go.mod h1:ac7W+fPG4pcb8yYp/npchGewPJezAu6UjlN0DjPEuH8=
2018
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
2119
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=

‎pkg/rsqueue/impls/database/groupqueue_test.go‎

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -103,9 +103,6 @@ func (f *fakeQueue) IsAddressInQueue(address string) (bool, error) {
103103
func (f *fakeQueue) PollAddress(address string) (errs <-chan error) {
104104
return f.pollErrs
105105
}
106-
func (f *fakeQueue) RecordFailure(address string, failure error) error {
107-
return f.record
108-
}
109106
func (f *fakeQueue) Get(maxPriority uint64, maxPriorityChan chan uint64, types queue.QueueSupportedTypes, stop chan bool) (*queue.QueueWork, error) {
110107
return nil, errors.New("n/i")
111108
}

‎pkg/rsqueue/impls/database/queue.go‎

Lines changed: 15 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -234,10 +234,6 @@ func (q *DatabaseQueue) AddressedPush(priority uint64, groupId int64, address st
234234
return err
235235
}
236236

237-
func (q *DatabaseQueue) RecordFailure(address string, failure error) error {
238-
return q.store.QueueAddressedComplete(address, failure)
239-
}
240-
241237
func (q *DatabaseQueue) IsAddressInQueue(address string) (bool, error) {
242238
return q.store.IsQueueAddressInProgress(address)
243239
}
@@ -247,22 +243,22 @@ func (q *DatabaseQueue) PollAddress(address string) <-chan error {
247243

248244
go func() {
249245
for {
250-
isDone, err := q.store.IsQueueAddressComplete(address)
246+
queued, err := q.IsAddressInQueue(address)
251247
if err != nil {
252248
// Ignore lock errors
253249
if !utils.IsSqliteLockError(err) {
254250
errCh <- err
255251
close(errCh)
256252
return
257253
}
258-
} else if isDone {
254+
} else if !queued {
259255
q.debugLogger.Debugf("Queue work with address %s completed", address)
260256
close(errCh)
261257
return
262258
}
263259

264260
// Wait for a notification or an interval, then poll again
265-
done := func() bool {
261+
done, err := func() (bool, error) {
266262
completedMsgs := q.SubscribeOne(q.notifyTypeWorkComplete, func(n listener.Notification) bool {
267263
if wn, ok := n.(*agenttypes.WorkCompleteNotification); ok {
268264
return wn.Address == address
@@ -279,20 +275,25 @@ func (q *DatabaseQueue) PollAddress(address string) <-chan error {
279275
defer tick.Stop()
280276
for {
281277
select {
282-
case <-completedMsgs:
278+
case n := <-completedMsgs:
283279
q.debugLogger.Debugf("Queue was notified that work with address %s completed", address)
284-
return false
280+
if wn, ok := n.(*agenttypes.WorkCompleteNotification); ok {
281+
return true, wn.Error
282+
}
285283
case <-chunkMsgs:
286284
q.debugLogger.Debugf("Queue was notified that chunk with address %s is ready", address)
287-
return true
285+
return true, nil
288286
case <-tick.C:
289-
return false
287+
return false, nil
290288
}
291289
}
292290
}()
293-
// If we received a chunk notification, then we return immediately so the client can
294-
// begin downloading chunks
291+
// If we received a chunk or work complete notification, then we return immediately so the client can
292+
// continue.
295293
if done {
294+
if err != nil {
295+
errCh <- err
296+
}
296297
close(errCh)
297298
return
298299
}
@@ -372,7 +373,7 @@ func (q *DatabaseQueue) Get(maxPriority uint64, maxPriorityChan chan uint64, typ
372373
if err != nil {
373374
return queueWork, err
374375
}
375-
queueWork, err := q.store.QueuePop(q.name, maxPriority, types.Enabled())
376+
queueWork, err = q.store.QueuePop(q.name, maxPriority, types.Enabled())
376377
if err != sql.ErrNoRows {
377378
q.measureDequeue(queueWork, err)
378379
return queueWork, err

0 commit comments

Comments
 (0)