gemma: pkg/imports/queue.go comparison

comparison pkg/imports/queue.go @ 1985:8eeb0b5eb340

Imports: Made retries and the waiting between the attempts configurable.

author	Sascha L. Teichmann <sascha.teichmann@intevation.de>
date	Wed, 23 Jan 2019 17:58:57 +0100
parents	d966f03ea819
children	59055c8301df

comparison

equal deleted inserted replaced

-:f9f1babe52ae
+:8eeb0b5eb340
 		Info(fmt string, args ...interface{})
 		// Warn logs warnings.
 		Warn(fmt string, args ...interface{})
 		// Error logs errors.
 		Error(fmt string, args ...interface{})
-	}
-	// RetryError is an error type to signal that
-	// the import should be tried again.
-	RetryError struct {
-		// Message is the error message.
-		Message string
-		// When is the new scheduled execution time.
-		When time.Time
 	}
 	// UnchangedError may be issued by Do of a Job to indicate
 	// That the database has not changed.
 	UnchangedError string
 	idJob struct {
 		id        int64
 		kind      JobKind
 		user      string
+		waitRetry pgtype.Interval
 		trysLeft  sql.NullInt64
 		sendEmail bool
 		data      string
 	}
 )
 	insertJobSQL = `
 INSERT INTO waterway.imports (
 kind,
 due,
 trys_left,
+retry_wait,
 username,
 send_email,
 data
 ) VALUES (
 $1,
 COALESCE($2, CURRENT_TIMESTAMP),
 $3,
 $4,
 $5,
-$6
+$6,
+$7
 ) RETURNING id`
 	selectJobSQL = `
 SELECT
 id,
 kind,
 trys_left,
+retry_wait,
 username,
 send_email,
 data
 FROM waterway.imports
 WHERE
 func init() {
 	go iqueue.importLoop()
 }
-// Error makes RetryError an error.
-func (re *RetryError) Error() string {
-	return re.Message
-}
 // Error makes UnchangedError an error.
 func (ue UnchangedError) Error() string {
 	return string(ue)
 }
 	}
 	// XXX: Consider using sort.Strings to make output deterministic.
 	return names
 }
-func (idj *idJob) trys() int {
+func (idj *idJob) nextRetry(feedback Feedback) bool {
+	switch {
+	case idj.waitRetry.Status != pgtype.Present && !idj.trysLeft.Valid:
+		return false
+	case idj.waitRetry.Status == pgtype.Present && !idj.trysLeft.Valid:
+		return true
+	case idj.trysLeft.Valid:
+		if idj.trysLeft.Int64 < 1 {
+			feedback.Warn("import should be retried, but no retrys left")
+		} else {
+			idj.trysLeft.Int64--
+			feedback.Info("import failed but will be retried")
+			return true
+		}
+	}
+	return false
+}
+func (idj *idJob) nextDue() time.Time {
+	now := time.Now()
+	if idj.waitRetry.Status == pgtype.Present {
+		var d time.Duration
+		if err := idj.waitRetry.AssignTo(&d); err != nil {
+			log.Printf("error: converting waitRetry failed: %v\n", err)
+		} else {
+			now = now.Add(d)
+		}
+	}
+	return now
+}
+func (idj *idJob) trysLeftPointer() *int {
 	if !idj.trysLeft.Valid {
-		return -1
+		return nil
 	}
-	return int(idj.trysLeft.Int64)
+	t := int(idj.trysLeft.Int64)
+	return &t
+}
+func (idj *idJob) waitRetryPointer() *time.Duration {
+	if idj.waitRetry.Status != pgtype.Present {
+		return nil
+	}
+	d := new(time.Duration)
+	if err := idj.waitRetry.AssignTo(d); err != nil {
+		log.Printf("error: converting waitRetry failed: %v\n", err)
+		return nil
+	}
+	return d
 }
 func (q *importQueue) jobCreator(kind JobKind) JobCreator {
 	q.creatorsMu.Lock()
 	defer q.creatorsMu.Unlock()
 }
 func (q *importQueue) addJob(
 	kind JobKind,
 	due time.Time,
-	trysLeft int,
+	trysLeft *int,
+	waitRetry *time.Duration,
 	user string,
 	sendEmail bool,
 	data string,
 ) (int64, error) {
-	ctx := context.Background()
 	var id int64
 	if due.IsZero() {
 		due = time.Now()
 	}
 	var tl sql.NullInt64
-	if trysLeft >= 0 {
+	if trysLeft != nil {
-		tl = sql.NullInt64{Int64: int64(trysLeft), Valid: true}
+		tl = sql.NullInt64{Int64: int64(*trysLeft), Valid: true}
 	}
+	var wr pgtype.Interval
+	if waitRetry != nil {
+		if err := wr.Set(*waitRetry); err != nil {
+			return 0, err
+		}
+	} else {
+		wr = pgtype.Interval{Status: pgtype.Null}
+	}
+	ctx := context.Background()
 	err := auth.RunAs(ctx, user, func(conn *sql.Conn) error {
 		return conn.QueryRowContext(
 			ctx,
 			insertJobSQL,
 			string(kind),
 			due,
 			tl,
+			&wr,
 			user,
 			sendEmail,
 			data).Scan(&id)
 	})
 	if err == nil {
 // This is gone in a separate Go routine
 // so this will not block.
 func AddJob(
 	kind JobKind,
 	due time.Time,
-	trysLeft int,
+	trysLeft *int,
+	waitRetry *time.Duration,
 	user string,
 	sendEmail bool,
 	data string,
 ) (int64, error) {
-	return iqueue.addJob(kind, due, trysLeft, user, sendEmail, data)
+	return iqueue.addJob(
+		kind,
+		due,
+		trysLeft,
+		waitRetry,
+		user,
+		sendEmail,
+		data)
 }
 type logFeedback int64
 func (lf logFeedback) log(kind, format string, args ...interface{}) {
 		defer tx.Rollback()
 		if err = tx.QueryRowContext(ctx, selectJobSQL, &kinds).Scan(
 			&ji.id,
 			&ji.kind,
 			&ji.trysLeft,
+			&ji.waitRetry,
 			&ji.user,
 			&ji.sendEmail,
 			&ji.data,
 		); err != nil {
 			return err
 						summary, err = job.Do(ctx, idj.id, conn, feedback)
 						return err
 					})
 			})()
-			var retry *RetryError
+			var unchanged, retry bool
-			var unchanged bool
+			if v, ok := errDo.(UnchangedError); ok {
-			switch v := errDo.(type) {
-			case *RetryError:
-				// NULL -> limit less
-				if idj.trysLeft.Valid && idj.trysLeft.Int64 <= 1 {
-					feedback.Warn("import should be retried, but no retrys left")
-				} else {
-					if idj.trysLeft.Valid {
-						idj.trysLeft.Int64--
-					}
-					feedback.Info("import failed but will be retried")
-					retry = v
-				}
-			case UnchangedError:
 				feedback.Info("unchanged: %s", v.Error())
 				unchanged = true
-			default:
+			} else if errDo != nil {
-				if errDo != nil {
+				feedback.Error("error in import: %v", errDo)
-					feedback.Error("error in import: %v", errDo)
+				retry = idj.nextRetry(feedback)
-				}
 			}
 			var errCleanup error
-			if retry == nil { // cleanup debris
+			if retry { // cleanup debris
 				if errCleanup = survive(job.CleanUp)(); errCleanup != nil {
 					feedback.Error("error cleanup: %v", errCleanup)
 				}
 			}
 			log.Printf("info: import #%d finished: %s\n", idj.id, state)
 			if idj.sendEmail {
 				go sendNotificationMail(idj.user, jc.Description(), state, idj.id)
 			}
-			if retry != nil {
+			if retry {
 				nid, err := q.addJob(
 					idj.kind,
-					retry.When, idj.trys(),
+					idj.nextDue(),
+					idj.trysLeftPointer(),
+					idj.waitRetryPointer(),
 					idj.user, idj.sendEmail,
 					idj.data)
 				if err != nil {
 					log.Printf("error: retry enqueue failed: %v\n", err)
 				} else {

Mercurial > gemma

comparison pkg/imports/queue.go @ 1985:8eeb0b5eb340