volume.DeleteBlob

Author: joonas-fi

frontend/pages/views/IntegrityVerificationJobsView.tsx

@@ -18,6 +18,7 @@ import {
 	IntegrityverificationjobClearProblems,
 	IntegrityverificationjobResume,
 	IntegrityverificationjobStop,
+	VolumeDeleteBlob,
 	VolumeVerifyIntegrity,
 } from 'generated/stoserver/stoservertypes_commands';
 import {
@@ -152,6 +153,14 @@ export default class IntegrityVerificationJobsView extends React.Component<
 			/>
 		);
 
+		const deleteBlob = (
+			<CommandLink
+				command={VolumeDeleteBlob(vol.Id, {
+					disambiguation: vol.Label,
+				})}
+			/>
+		);
+
 		const volumeMounted = this.props.mounts.some((m) => m.Volume === vol.Id && m.Online);
 
 		return (
@@ -199,6 +208,7 @@ export default class IntegrityVerificationJobsView extends React.Component<
 					{completed && (
 						<Dropdown>
 							{startJob}
+							{deleteBlob}
 							{job.ErrorsFound > 0 && (
 								<CommandLink
 									command={IntegrityverificationjobClearProblems(job.Id)}
@@ -211,6 +221,7 @@ export default class IntegrityVerificationJobsView extends React.Component<
 							<CommandLink command={IntegrityverificationjobResume(job.Id)} />
 							<CommandLink command={IntegrityverificationjobStop(job.Id)} />
 							{startJob}
+							{deleteBlob}
 						</Dropdown>
 					)}
 				</td>

pkg/stoserver/commandhandlersdatamanagement.go

@@ -3,6 +3,7 @@ package stoserver
 import (
 	"errors"
 	"fmt"
+	"log/slog"
 	"slices"
 	"time"
 
@@ -210,6 +211,53 @@ func (c *cHandlers) VolumeMarkDataLost(cmd *stoservertypes.VolumeMarkDataLost, c
 	})
 }
 
+func (c *cHandlers) VolumeDeleteBlob(cmd *stoservertypes.VolumeDeleteBlob, ctx *command.Ctx) error {
+	ref, err := stotypes.BlobRefFromHex(cmd.Ref)
+	if err != nil {
+		return err
+	}
+
+	return c.db.Update(func(tx *bbolt.Tx) error {
+		volToRemoveFrom, err := stodb.Read(tx).Volume(cmd.Id)
+		if err != nil {
+			return err
+		}
+
+		blob, err := stodb.Read(tx).Blob(*ref)
+		if err != nil {
+			return err
+		}
+
+		volumeRemoved := lo.Filter(blob.Volumes, func(volID int, _ int) bool { return volID != volToRemoveFrom.ID })
+
+		if removalHasNoChange := len(volumeRemoved) == len(blob.Volumes); removalHasNoChange {
+			return fmt.Errorf("requested blob %s not found in volume %s", cmd.Ref, volToRemoveFrom.Label)
+		}
+
+		if err := c.conf.DiskAccess.Delete(ctx.Ctx, *ref, volToRemoveFrom.ID); err != nil {
+			// don't abort because blob may be missing in storage and we must still be able to do book-keeping changes
+			// i.e. mark in metadata DB that volume no longer has blob.
+			slog.Warn("failed to remove from underlying storage", "blob", ref.AsHex())
+		}
+
+		volToRemoveFrom.BlobSizeTotal -= int64(blob.SizeOnDisk)
+		volToRemoveFrom.BlobCount--
+
+		blob.Volumes = volumeRemoved
+
+		if err := stodb.VolumeRepository.Update(volToRemoveFrom, tx); err != nil {
+			return err
+		}
+
+		if err := stodb.BlobRepository.Update(blob, tx); err != nil {
+			return err
+		}
+
+		return nil
+	})
+}
+
+// even though collections are given as argument, only the blobs with problems are replicated to the target volume
 func (c *cHandlers) DatabaseReconcileReplicationPolicy(cmd *stoservertypes.DatabaseReconcileReplicationPolicy, ctx *command.Ctx) error {
 	collIDs := *cmd.Collections
 

pkg/stoserver/stodiskaccess/diskaccess.go

@@ -268,6 +268,23 @@ func (d *Controller) BestVolumeID(volumeIDs []int) (int, error) {
 	return lowestCostVolumeID, nil
 }
 
+// deletes a blob from one volume
+func (d *Controller) Delete(ctx context.Context, ref stotypes.BlobRef, volumeID int) error {
+	driver, err := d.driverFor(volumeID)
+	if err != nil {
+		return err
+	}
+
+	if err := driver.RawDelete(ctx, ref); err != nil {
+		return err
+	}
+
+	// NOTE: not calling calling metadata store to record blob having been removed. right now we leave it to caller
+	// in order for the caller (to let it itself decide on what to do with errors removing the blob from underlying storage)
+
+	return nil
+}
+
 // runs a scrub for a blob in a given volume to detect errors
 // https://en.wikipedia.org/wiki/Data_scrubbing
 // we could actually just do a Fetch() but that would require access to the encryption keys.

pkg/stoserver/stodiskaccess/diskaccess_test.go

@@ -432,6 +432,14 @@ func (t *testingBlobStorage) RawStore(_ context.Context, ref stotypes.BlobRef, c
 	return nil
 }
 
+func (t *testingBlobStorage) RawDelete(ctx context.Context, ref stotypes.BlobRef) error {
+	if _, has := t.files[ref.AsHex()]; !has {
+		return os.ErrNotExist
+	}
+	delete(t.files, ref.AsHex())
+	return nil
+}
+
 func xorSlices(a []byte, b []byte) []byte {
 	if len(a) != len(b) {
 		panic("nope")

pkg/stoserver/stoservertypes/commands.json

@@ -261,6 +261,17 @@
 			{ "key": "OnlyIfRedundancy", "title": "Abort if blob would lose last replica", "type": "checkbox", "help": "Failsafe for you to reconsider if it losing last replica would come as a surprise." }
 		]
 	},
+	{
+		"command": "volume.DeleteBlob",
+		"chain": "public",
+		"ctor": ["Id"],
+		"crudNature": "delete",
+		"title": "Delete blob (or mark corrupted one lost)",
+		"fields": [
+			{ "key": "Id", "type": "integer", "hideIfDefaultValue": true },
+			{ "key": "Ref", "title": "Blob reference" }
+		]
+	},
 	{
 		"command": "volume.Decommission",
 		"chain": "public",