feat: implement garbage collection of old operations

internal/oplog/oplog.go

@@ -248,11 +248,8 @@ func (o *OpLog) nextOperationId(b *bolt.Bucket, unixTimeMs int64) (int64, error)
 func (o *OpLog) addOperationHelper(tx *bolt.Tx, op *v1.Operation) error {
 	b := tx.Bucket(OpLogBucket)
 	if op.Id == 0 {
-		if op.UnixTimeStartMs == 0 {
-			return fmt.Errorf("operation must have a start time")
-		}
 		var err error
-		op.Id, err = o.nextOperationId(b, op.UnixTimeStartMs)
+		op.Id, err = o.nextOperationId(b, time.Now().UnixMilli())
 		if err != nil {
 			return fmt.Errorf("create next operation ID: %w", err)
 		}

internal/orchestrator/orchestrator.go

@@ -127,6 +127,9 @@ func (o *Orchestrator) ApplyConfig(cfg *v1.Config) error {
 	zap.L().Info("Applied config to orchestrator, task queue reset. Rescheduling planned tasks now.")
 
 	// Requeue tasks that are affected by the config change.
+	o.ScheduleTask(&CollectGarbageTask{
+		orchestrator: o,
+	}, TaskPriorityDefault)
 	for _, plan := range cfg.Plans {
 		t, err := NewScheduledBackupTask(o, plan)
 		if err != nil {

internal/orchestrator/taskcollectgarbage.go

@@ -0,0 +1,118 @@
+package orchestrator
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	v1 "github.com/garethgeorge/restora/gen/go/v1"
+	"go.uber.org/zap"
+)
+
+const (
+	gcStartupDelay = 5 * time.Second
+	gcInterval     = 24 * time.Hour
+	// keep operations that are eligible for gc for 30 days OR up to a limit of 100 for any one plan.
+	// an operation is eligible for gc if:
+	// - it has no snapshot associated with it
+	// - it has a forgotten snapshot associated with it
+	gcHistoryAge      = 30 * 24 * time.Hour
+	gcHistoryMaxCount = 100
+)
+
+type CollectGarbageTask struct {
+	orchestrator *Orchestrator // owning orchestrator
+	firstRun     bool
+}
+
+var _ Task = &CollectGarbageTask{}
+
+func (t *CollectGarbageTask) Name() string {
+	return "collect garbage"
+}
+
+func (t *CollectGarbageTask) Next(now time.Time) *time.Time {
+	if !t.firstRun {
+		t.firstRun = true
+		runAt := now.Add(gcStartupDelay)
+		return &runAt
+	}
+
+	runAt := now.Add(gcInterval)
+	return &runAt
+}
+
+func (t *CollectGarbageTask) Run(ctx context.Context) error {
+	oplog := t.orchestrator.OpLog
+
+	// pass 1: identify forgotten snapshots.
+	snapshotIsForgotten := make(map[string]bool)
+	if err := oplog.ForAll(func(op *v1.Operation) error {
+		if snapshotOp, ok := op.Op.(*v1.Operation_OperationIndexSnapshot); ok {
+			if snapshotOp.OperationIndexSnapshot.Forgot {
+				snapshotIsForgotten[snapshotOp.OperationIndexSnapshot.Snapshot.Id] = true
+			}
+		}
+		return nil
+	}); err != nil {
+		return fmt.Errorf("identifying forgotten snapshots: %w", err)
+	}
+
+	// pass 2: identify operations that are gc eligible
+	//  - any operation that has no snapshot associated with it
+	//  - any operation that has a forgotten snapshot associated with it
+	operationsByPlan := make(map[string][]gcOpInfo)
+	if err := oplog.ForAll(func(op *v1.Operation) error {
+		if op.SnapshotId == "" || snapshotIsForgotten[op.SnapshotId] {
+			operationsByPlan[op.PlanId] = append(operationsByPlan[op.PlanId], gcOpInfo{
+				id:        op.Id,
+				timestamp: op.UnixTimeStartMs,
+			})
+		}
+		return nil
+	}); err != nil {
+		return fmt.Errorf("identifying gc eligible operations: %w", err)
+	}
+
+	var gcOps []int64
+	curTime := curTimeMillis()
+	for _, opInfos := range operationsByPlan {
+		if len(opInfos) >= gcHistoryMaxCount {
+			for _, opInfo := range opInfos[:len(opInfos)-gcHistoryMaxCount] {
+				gcOps = append(gcOps, opInfo.id)
+			}
+			opInfos = opInfos[len(opInfos)-gcHistoryMaxCount:]
+		}
+
+		// check if each operation timestamp is old.
+		for _, opInfo := range opInfos {
+			if curTime-opInfo.timestamp > gcHistoryAge.Milliseconds() {
+				gcOps = append(gcOps, opInfo.id)
+			}
+		}
+	}
+
+	// pass 3: remove gc eligible operations
+	if err := oplog.Delete(gcOps...); err != nil {
+		return fmt.Errorf("removing gc eligible operations: %w", err)
+	}
+
+	zap.L().Info("collecting garbage",
+		zap.Int("forgotten_snapshots", len(snapshotIsForgotten)),
+		zap.Any("operations_removed", len(gcOps)))
+
+	return nil
+}
+
+func (t *CollectGarbageTask) Cancel(withStatus v1.OperationStatus) error {
+	return nil
+}
+
+func (t *CollectGarbageTask) OperationId() int64 {
+	return 0
+}
+
+type gcOpInfo struct {
+	id        int64 // operation ID
+	timestamp int64 // unix time milliseconds
+}

internal/orchestrator/taskforget.go

@@ -99,10 +99,6 @@ func (t *ForgetTask) Run(ctx context.Context) error {
 					continue
 				}
 			}
-			// Soft delete the operation (can be recovered if necessary, todo: implement recovery).
-			if e := t.orch.OpLog.Delete(op.Id); err != nil {
-				err = multierror.Append(err, fmt.Errorf("delete operation %v: %w", op.Id, e))
-			}
 		}
 
 		if len(forgot) > 0 {

proto/v1/operations.proto

@@ -12,7 +12,7 @@ message OperationList {
 }
 
 message Operation {
-  // required, primary ID of the operation.
+  // required, primary ID of the operation. ID is sequential based on creation time of the operation.
   int64 id = 1;
   // required, repo id if associated with a repo
   string repo_id = 2; 

webui/src/components/ActivityBar.tsx

@@ -35,7 +35,7 @@ export const ActivityBar = () => {
         }
     });
 
-    return <span>{details.map(details => {
-        return <>{details.displayName} in progress for plan {details.op.planId} to {details.op.repoId} for {formatDuration(details.details.duration)}</>
+    return <span>{details.map((details, idx) => {
+        return <span key={idx}>{details.displayName} in progress for plan {details.op.planId} to {details.op.repoId} for {formatDuration(details.details.duration)}</span>
     })}</span>
 }

webui/src/components/OperationList.tsx

@@ -86,13 +86,13 @@ export const OperationList = ({
       };
       subscribeToOperations(lis);
 
-      backupCollector.subscribe(() => {
+      backupCollector.subscribe(_.debounce(() => {
         let backups = backupCollector.getAll();
         backups.sort((a, b) => {
           return b.startTimeMs - a.startTimeMs;
         });
         setBackups(backups);
-      });
+      }, 50));
 
       getOperations(req)
         .then((ops) => {
@@ -106,7 +106,10 @@ export const OperationList = ({
       };
     }, [JSON.stringify(req)]);
   } else {
-    backups = useBackups || [];
+    backups = [...(useBackups || [])];
+    backups.sort((a, b) => {
+      return b.startTimeMs - a.startTimeMs;
+    });
   }
 
   if (backups.length === 0) {
@@ -277,14 +280,14 @@ export const OperationRow = ({
               children: <>
                 Removed snapshots:
                 <pre>{forgetOp.forget?.map((f) => (
-                  <>
+                  <div key={f.id}>
                     {"removed snapshot " + normalizeSnapshotId(f.id!) + " taken at " + formatTime(f.unixTimeMs!)} <br />
-                  </>
+                  </div>
                 ))}</pre>
                 Policy:
                 <ul>
-                  {policyDesc.map((desc) => (
-                    <li>{desc}</li>
+                  {policyDesc.map((desc, idx) => (
+                    <li key={idx}>{desc}</li>
                   ))}
                 </ul>
               </>,

webui/src/components/OperationTree.tsx

@@ -58,13 +58,13 @@ export const OperationTree = ({
     };
     subscribeToOperations(lis);
 
-    backupCollector.subscribe(() => {
+    backupCollector.subscribe(_.debounce(() => {
       let backups = backupCollector.getAll();
       backups.sort((a, b) => {
         return b.startTimeMs - a.startTimeMs;
       });
       setBackups(backups);
-    });
+    }, 50));
 
     getOperations(req)
       .then((ops) => {
@@ -79,7 +79,7 @@ export const OperationTree = ({
   }, [JSON.stringify(req)]);
 
   const treeData = useMemo(() => {
-    return buildTreeYear(backups);
+    return buildTreePlan(backups);
   }, [backups]);
 
   if (backups.length === 0) {
@@ -199,6 +199,26 @@ export const OperationTree = ({
   );
 };
 
+const buildTreePlan = (operations: BackupInfo[]): OpTreeNode[] => {
+  const grouped = _.groupBy(operations, (op) => {
+    return op.planId;
+  });
+
+  const entries: OpTreeNode[] = _.map(grouped, (value, key) => {
+    return {
+      key: "p" + key,
+      title: "" + key,
+      children: buildTreeYear(value),
+    };
+  });
+  entries.sort(sortByKey);
+
+  if (entries.length === 1) {
+    return entries[0].children!;
+  }
+  return entries;
+};
+
 const buildTreeYear = (operations: BackupInfo[]): OpTreeNode[] => {
   const grouped = _.groupBy(operations, (op) => {
     return localISOTime(op.displayTime).substring(0, 4);

webui/src/components/SnapshotBrowser.tsx

@@ -1,5 +1,5 @@
 import React, { useEffect, useMemo, useState } from "react";
-import { Button, Dropdown, Form, Input, Modal, Space, Tree } from "antd";
+import { Button, Dropdown, Form, Input, Modal, Space, Spin, Tree } from "antd";
 import type { DataNode, EventDataNode } from "antd/es/tree";
 import {
   ListSnapshotFilesResponse,
@@ -109,7 +109,7 @@ export const SnapshotBrowser = ({
       },
       { pathPrefix: "/api" }
     );
-    
+
     setTreeData((treeData) => {
       let toUpdate: DataNode | null = null;
       for (const node of treeData) {
@@ -118,14 +118,14 @@ export const SnapshotBrowser = ({
           break;
         }
       }
-  
+
       if (!toUpdate) {
         return treeData;
       }
-  
+
       const toUpdateCopy = { ...toUpdate };
       toUpdateCopy.children = respToNodes(resp);
-  
+
       return treeData.map((node) => {
         const didUpdate = replaceKeyInTree(node, key as string, toUpdateCopy);
         if (didUpdate) {
@@ -136,6 +136,10 @@ export const SnapshotBrowser = ({
     });
   };
 
+  if (treeData.length === 0) {
+    return <Spin />;
+  }
+
   return (
     <SnapshotBrowserContext.Provider
       value={{ snapshotId, repoId, planId, showModal }}
@@ -327,4 +331,4 @@ const RestoreModal = ({
   );
 };
 
-const restoreFlow = (repoId: string, snapshotId: string, path: string) => {};
+const restoreFlow = (repoId: string, snapshotId: string, path: string) => { };

webui/src/state/oplog.ts

@@ -94,12 +94,13 @@ export interface BackupInfo {
   startTimeMs: number;
   endTimeMs: number;
   status: OperationStatus;
-  operations: Operation[];
+  operations: Operation[]; // operations ordered by their unixTimeStartMs (not ID)
   repoId?: string;
   planId?: string;
   snapshotId?: string;
   backupLastStatus?: BackupProgressEntry;
   snapshotInfo?: ResticSnapshot;
+  forgotten: boolean;
 }
 
 // BackupInfoCollector maps multiple operations to single aggregate 'BackupInfo' objects.
@@ -112,31 +113,57 @@ export class BackupInfoCollector {
   private backupByOpId: { [key: string]: BackupInfo } = {};
   private backupBySnapshotId: { [key: string]: BackupInfo } = {};
 
-  private mergeBackups(existing: BackupInfo, newInfo: BackupInfo) {
-    if (existing.id > newInfo.id) {
-      existing.id = newInfo.id;
-    }
-    existing.startTimeMs = Math.min(existing.startTimeMs, newInfo.startTimeMs);
-    existing.endTimeMs = Math.max(existing.endTimeMs, newInfo.endTimeMs);
-    existing.displayTime = new Date(existing.startTimeMs);
-    existing.displayType = DisplayType.SNAPSHOT;
-    if (newInfo.startTimeMs >= existing.startTimeMs && newInfo.status !== OperationStatus.STATUS_SYSTEM_CANCELLED) { // don't overwrite with cancelled status since that operation will be hidden.
-      existing.status = newInfo.status; // use the latest status
-    }
-    existing.operations = _.uniqBy(
-      [...newInfo.operations, ...existing.operations],
-      (o) => o.id!
-    );
-    existing.operations.sort((a, b) => {
+  private createBackup(operations: Operation[]): BackupInfo {
+    // deduplicate and sort operations.
+    operations.sort((a, b) => {
       return parseInt(b.unixTimeStartMs!) - parseInt(a.unixTimeStartMs!);
     });
-    if (newInfo.backupLastStatus) {
-      existing.backupLastStatus = newInfo.backupLastStatus;
+
+    // use the lowest ID of all operations as the ID of the backup, this will be the first created operation.
+    const id = operations.reduce((prev, curr) => {
+      return prev < curr.id! ? prev : curr.id!;
+    }, operations[0].id!);
+
+    const startTimeMs = parseInt(operations[0].unixTimeStartMs!);
+    const endTimeMs = parseInt(operations[operations.length - 1].unixTimeEndMs!);
+    const displayTime = new Date(startTimeMs);
+    let displayType = DisplayType.SNAPSHOT;
+    if (operations.length === 1) {
+      displayType = getTypeForDisplay(operations[0]);
     }
-    if (newInfo.snapshotInfo) {
-      existing.snapshotInfo = newInfo.snapshotInfo;
+
+    // use the latest status that is not cancelled.
+    let statusIdx = operations.length - 1;
+    let status = OperationStatus.STATUS_SYSTEM_CANCELLED;
+    while (statusIdx > 0 || shouldHideStatus(status)) {
+      status = operations[statusIdx].status!;
+      statusIdx--;
     }
-    return existing;
+
+    let backupLastStatus = undefined;
+    let snapshotInfo = undefined;
+    let forgotten = false;
+    for (const op of operations) {
+      if (op.operationBackup) {
+        backupLastStatus = op.operationBackup.lastStatus;
+      } else if (op.operationIndexSnapshot) {
+        snapshotInfo = op.operationIndexSnapshot.snapshot;
+        forgotten = op.operationIndexSnapshot.forgot || false;
+      }
+    }
+
+    return {
+      id,
+      startTimeMs,
+      endTimeMs,
+      displayTime,
+      displayType,
+      status,
+      operations,
+      backupLastStatus,
+      snapshotInfo,
+      forgotten,
+    };
   }
 
   private operationToBackup(op: Operation): BackupInfo {
@@ -167,21 +194,29 @@ export class BackupInfoCollector {
     return b;
   }
 
-  private addHelper(op: Operation) {
+  private addHelper(op: Operation): BackupInfo {
     if (op.snapshotId) {
       delete this.backupByOpId[op.id!];
 
-      let newInfo = this.operationToBackup(op);
       const existing = this.backupBySnapshotId[op.snapshotId!];
+      let operations: Operation[];
       if (existing) {
-        this.mergeBackups(existing, newInfo);
-        return existing;
+        operations = [...existing.operations];
+        const opIdx = operations.findIndex((o) => o.id === op.id);
+        if (opIdx > -1) {
+          operations[opIdx] = op;
+        } else {
+          operations.push(op);
+        }
       } else {
-        this.backupBySnapshotId[op.snapshotId] = newInfo;
-        return newInfo;
+        operations = [op];
       }
+
+      const newInfo = this.createBackup(operations);
+      this.backupBySnapshotId[op.snapshotId!] = newInfo;
+      return newInfo;
     } else {
-      const newInfo = this.operationToBackup(op);
+      const newInfo = this.createBackup([op]);
       this.backupByOpId[op.id!] = newInfo;
       return newInfo;
     }
@@ -212,15 +247,11 @@ export class BackupInfoCollector {
     return backupInfos;
   }
 
-  public addOperationNoNotify(op: Operation) {
-    this.addHelper(op);
-  }
-
   public getAll(): BackupInfo[] {
     const arr = [];
     arr.push(...Object.values(this.backupByOpId));
     arr.push(...Object.values(this.backupBySnapshotId));
-    return arr;
+    return arr.filter((b) => !b.forgotten && !shouldHideStatus(b.status));
   }
 
   public subscribe(

webui/src/views/AddRepoModal.tsx

@@ -77,9 +77,9 @@ export const AddRepoModal = ({
       showModal(null);
       alertsApi.success(
         "Deleted repo " +
-          template.id +
-          " from config but files remain. To release storage delete the files manually. URI: " +
-          template.uri
+        template.id +
+        " from config but files remain. To release storage delete the files manually. URI: " +
+        template.uri
       );
     } catch (e: any) {
       alertsApi.error("Operation failed: " + e.message, 15);

webui/src/views/App.tsx

@@ -68,7 +68,6 @@ export const App: React.FC = () => {
           <small style={{ color: "rgba(255,255,255,0.3)", fontSize: "0.6em" }}>
             {uiBuildVersion}
           </small>
-
           <small style={{ fontSize: "0.6em", marginLeft: "30px" }}>
             <ActivityBar />
           </small>

webui/src/views/PlanView.tsx

@@ -69,7 +69,7 @@ export const PlanView = ({ plan }: React.PropsWithChildren<{ plan: Plan }>) => {
     <>
       <Flex gap="small" align="center" wrap="wrap">
         <Typography.Title>
-          <h1>{plan.id}</h1>
+          {plan.id}
         </Typography.Title>
       </Flex>
       <Flex gap="small" align="center" wrap="wrap">