Fixed task partition cleanup (#61)

Author: djthorpe

pgqueue/manager/partition.go

@@ -102,6 +102,16 @@ func (manager *Manager) DropDrainedPartition(ctx context.Context) (result string
 	if oldest.Count > 0 {
 		return "", nil
 	}
+
+	// Don't drop a partition if the sequence hasn't advanced past its end — new
+	// inserts could still land in this range.
+	seq, err := manager.GetPartitionSeq(ctx)
+	if err != nil {
+		return "", err
+	}
+	if seq < oldest.End {
+		return "", nil
+	}
 	if err := manager.DeletePartition(ctx, oldest.Partition); err != nil {
 		return "", err
 	} else {

queries.go

@@ -83,6 +83,11 @@ func NewQueries(r io.Reader) (*Queries, error) {
 		self.keys = append(self.keys, key)
 	}
 
+	// Remove any ; at the end of each SQL statement
+	for k, v := range self.queries {
+		self.queries[k] = strings.TrimSuffix(v, ";")
+	}
+
 	// Check for scanner errors
 	if err := scanner.Err(); err != nil {
 		return nil, err