Merge pull request #24 from ibuildthecloud/master

Dedup compact_rev_key on start

Author: ibuildthecloud

pkg/drivers/generic/generic.go

@@ -255,18 +255,7 @@ func (d *Generic) GetCompactRevision(ctx context.Context) (int64, error) {
 }
 
 func (d *Generic) SetCompactRevision(ctx context.Context, revision int64) error {
-	result, err := d.execute(ctx, d.UpdateCompactSQL, revision)
-	if err != nil {
-		return err
-	}
-	num, err := result.RowsAffected()
-	if err != nil {
-		return err
-	}
-	if num != 0 {
-		return nil
-	}
-	_, err = d.Insert(ctx, "compact_rev_key", false, false, 0, revision, 0, []byte(""), nil)
+	_, err := d.execute(ctx, d.UpdateCompactSQL, revision)
 	return err
 }
 

pkg/logstructured/sqllog/sql.go

@@ -46,6 +46,52 @@ func (s *SQLLog) Start(ctx context.Context) (err error) {
 	return
 }
 
+func (s *SQLLog) compactStart(ctx context.Context) error {
+	rows, err := s.d.After(ctx, "compact_rev_key", 0, 0)
+	if err != nil {
+		return err
+	}
+
+	_, _, events, err := RowsToEvents(rows)
+	if err != nil {
+		return err
+	}
+
+	if len(events) == 0 {
+		_, err := s.Append(ctx, &server.Event{
+			Create: true,
+			KV: &server.KeyValue{
+				Key:   "compact_rev_key",
+				Value: []byte(""),
+			},
+		})
+		return err
+	} else if len(events) == 1 {
+		return nil
+	}
+
+	// this is to work around a bug in which we ended up with two compact_rev_key rows
+	maxRev := int64(0)
+	maxID := int64(0)
+	for _, event := range events {
+		if event.PrevKV != nil && event.PrevKV.ModRevision > maxRev {
+			maxRev = event.PrevKV.ModRevision
+			maxID = event.KV.ModRevision
+		}
+	}
+
+	for _, event := range events {
+		if event.KV.ModRevision == maxID {
+			continue
+		}
+		if err := s.d.DeleteRevision(ctx, event.KV.ModRevision); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
 func (s *SQLLog) compact() {
 	var (
 		nextEnd int64
@@ -277,6 +323,10 @@ func filter(events interface{}, checkPrefix bool, prefix string) ([]*server.Even
 }
 
 func (s *SQLLog) startWatch() (chan interface{}, error) {
+	if err := s.compactStart(s.ctx); err != nil {
+		return nil, err
+	}
+
 	pollStart, err := s.d.GetCompactRevision(s.ctx)
 	if err != nil {
 		return nil, err