unacknowledged and incomplete scans are re-published to the notificat…

…ion queue

database/messaging.go

@@ -11,6 +11,9 @@ import (
 	"github.com/mozilla/tls-observatory/logger"
 )
 
+// RegisterScanListener "subscribes" to the notifications published to the scan_listener notifier.
+// It has as input the usual sb attributes and returns an int64 channel which can be consumed
+// for newly created scan id's.
 func (db *DB) RegisterScanListener(dbname, user, password, hostport, sslmode string) <-chan int64 {
 
 	log := logger.GetLogger()
@@ -24,20 +27,19 @@ func (db *DB) RegisterScanListener(dbname, user, password, hostport, sslmode str
 	}
 
 	listenerChan := make(chan int64)
+	listenerName := "scan_listener"
 
-	conn_info := fmt.Sprintf("postgres://%s:%s@%s/%s?sslmode=%s",
+	connInfo := fmt.Sprintf("postgres://%s:%s@%s/%s?sslmode=%s",
 		user, password, hostport, dbname, sslmode)
 
 	go func() {
 
-		listener_name := "scan_listener"
-
-		listener := pq.NewListener(conn_info, 100*time.Millisecond, 10*time.Second, reportProblem)
-		err := listener.Listen(listener_name)
+		listener := pq.NewListener(connInfo, 100*time.Millisecond, 10*time.Second, reportProblem)
+		err := listener.Listen(listenerName)
 
 		if err != nil {
 			log.WithFields(logrus.Fields{
-				"listener": listener_name,
+				"listener": listenerName,
 				"error":    err.Error(),
 			}).Error("could not listen for notification")
 			close(listenerChan)
@@ -66,6 +68,30 @@ func (db *DB) RegisterScanListener(dbname, user, password, hostport, sslmode str
 
 	}()
 
+	go func() {
+		fiveminticker := time.NewTicker(1 * time.Minute)
+		unackedQuery := fmt.Sprintf("select pg_notify('%s', ''||id ) from scans where ack=FALSE and timestamp < NOW() - INTERVAL '30 seconds'", listenerName)
+		zerocomplQuery := "update scans set ack=false where completion_perc=0 and timestamp < NOW() - INTERVAL '1 minute'"
+		for {
+			select {
+			case <-fiveminticker.C:
+				_, err := db.Exec(zerocomplQuery)
+				if err != nil {
+					log.WithFields(logrus.Fields{
+						"error": err,
+					}).Error("Could not run zero completion update query")
+				}
+
+				_, err = db.Exec(unackedQuery)
+				if err != nil {
+					log.WithFields(logrus.Fields{
+						"error": err,
+					}).Error("Could not run unacknowledged scans periodic check.")
+				}
+			}
+		}
+	}()
+
 	return listenerChan
 }