internal/server/store/volumes.go
Ref: Size: 11.3 KiB History
package store
import (
"database/sql"
"errors"
"fmt"
"time"
"github.com/a73x/eitri/internal/random"
sqlite "modernc.org/sqlite"
)
// VolumeClaim is a tenant's request for durable storage. HostID and VMID are
// joined facts: where the bytes were placed, and which VM holds them now.
type VolumeClaim struct {
ID, Tenant, Name string
SizeGB int64
BoundVolumeID string
HostID string
VMID string
CreatedAt time.Time
DeletedAt *time.Time
}
// Volume is the fleet's placement of one claim's bytes on one host.
type Volume struct {
ID, HostID, ClaimID string
SizeGB int64
CreatedAt time.Time
DeletedAt *time.Time
}
var (
ErrClaimNotFound = errors.New("volume claim not found")
ErrClaimNameTaken = errors.New("volume claim name already in use")
ErrClaimAttached = errors.New("volume claim is attached to a vm")
ErrClaimPinned = errors.New("volume claim is bound to another host")
)
// ClaimAttachedError names the VM that holds the claim.
type ClaimAttachedError struct{ ClaimID, VMID string }
func (e *ClaimAttachedError) Error() string {
return fmt.Sprintf("volume claim %s is attached to vm %s", e.ClaimID, e.VMID)
}
func (e *ClaimAttachedError) Is(target error) bool { return target == ErrClaimAttached }
// ClaimPinnedError names the host the claim's data lives on.
type ClaimPinnedError struct{ ClaimID, HostID string }
func (e *ClaimPinnedError) Error() string {
return fmt.Sprintf("volume claim %s is bound to host %s", e.ClaimID, e.HostID)
}
func (e *ClaimPinnedError) Is(target error) bool { return target == ErrClaimPinned }
// claimColumns / claimFrom are one positional contract with scanClaim.
const claimColumns = `c.id, c.tenant, c.name, c.size_gb, COALESCE(c.bound_volume_id,''),
COALESCE(v.host_id,''), COALESCE(a.vm_id,''), c.created_at, c.deleted_at`
const claimFrom = ` FROM volume_claims c
LEFT JOIN volumes v ON v.id = c.bound_volume_id
LEFT JOIN volume_attachments a ON a.claim_id = c.id`
// scanner is satisfied by *sql.Row and *sql.Rows alike, so the single-row read
// and the list share one scan.
type scanner interface{ Scan(dest ...any) error }
func scanClaim(row scanner) (VolumeClaim, error) {
var c VolumeClaim
var createdAt string
var deletedAt sql.NullString
if err := row.Scan(&c.ID, &c.Tenant, &c.Name, &c.SizeGB, &c.BoundVolumeID, &c.HostID, &c.VMID, &createdAt, &deletedAt); err != nil {
return VolumeClaim{}, err
}
c.CreatedAt, _ = time.Parse(time.RFC3339, createdAt)
if deletedAt.Valid {
t, _ := time.Parse(time.RFC3339, deletedAt.String)
c.DeletedAt = &t
}
return c, nil
}
// CreateVolumeClaim records a request for storage. It places nothing: a claim
// is Pending until the first VM that names it decides which host it lands on.
func (s *Store) CreateVolumeClaim(tenant, name string, sizeGB int64) (VolumeClaim, error) {
c := VolumeClaim{ID: random.Hex(16), Tenant: tenant, Name: name, SizeGB: sizeGB, CreatedAt: time.Now().UTC()}
_, err := s.db.Exec(`INSERT INTO volume_claims(id, tenant, name, size_gb, created_at) VALUES (?,?,?,?,?)`,
c.ID, c.Tenant, c.Name, c.SizeGB, c.CreatedAt.Format(time.RFC3339))
if err != nil {
// SQLITE_CONSTRAINT_UNIQUE (2067): volume_claims_tenant_name.
if serr, ok := errors.AsType[*sqlite.Error](err); ok && serr.Code() == 2067 {
return VolumeClaim{}, ErrClaimNameTaken
}
return VolumeClaim{}, fmt.Errorf("insert volume claim: %w", err)
}
return c, nil
}
// ListVolumeClaims returns tenant's live claims, oldest first.
func (s *Store) ListVolumeClaims(tenant string) ([]VolumeClaim, error) {
rows, err := s.db.Query(`SELECT `+claimColumns+claimFrom+` WHERE c.tenant=? AND c.deleted_at IS NULL ORDER BY c.created_at, c.id`, tenant)
if err != nil {
return nil, err
}
defer rows.Close()
var out []VolumeClaim
for rows.Next() {
c, err := scanClaim(rows)
if err != nil {
return nil, err
}
out = append(out, c)
}
return out, rows.Err()
}
// GetVolumeClaim reads one live claim; a tombstoned one reads as sql.ErrNoRows.
func (s *Store) GetVolumeClaim(id string) (VolumeClaim, error) {
return scanClaim(s.db.QueryRow(`SELECT `+claimColumns+claimFrom+` WHERE c.id=? AND c.deleted_at IS NULL`, id))
}
// GetVolumeClaimAny reads one claim whether live or tombstoned. It exists for
// the reap, which asks after a claim it is about to delete precisely because
// that claim is already tombstoned: GetVolumeClaim would answer ErrNoRows and
// the terminal audit row would lose its tenant.
func (s *Store) GetVolumeClaimAny(id string) (VolumeClaim, error) {
return scanClaim(s.db.QueryRow(`SELECT `+claimColumns+claimFrom+` WHERE c.id=?`, id))
}
// TombstoneVolumeClaim marks a claim and its bound volume (if any) for
// reclaim. Refused while a VM holds the claim: the guest may be writing.
func (s *Store) TombstoneVolumeClaim(id string) error {
tx, err := s.db.Begin()
if err != nil {
return err
}
defer tx.Rollback()
var vmID, bound sql.NullString
err = tx.QueryRow(`SELECT a.vm_id, c.bound_volume_id FROM volume_claims c
LEFT JOIN volume_attachments a ON a.claim_id=c.id WHERE c.id=? AND c.deleted_at IS NULL`, id).Scan(&vmID, &bound)
if err != nil {
return err // sql.ErrNoRows passes through
}
if vmID.Valid {
return &ClaimAttachedError{ClaimID: id, VMID: vmID.String}
}
now := time.Now().UTC().Format(time.RFC3339)
if _, err := tx.Exec(`UPDATE volume_claims SET deleted_at=? WHERE id=?`, now, id); err != nil {
return err
}
if bound.Valid {
if _, err := tx.Exec(`UPDATE volumes SET deleted_at=? WHERE id=? AND deleted_at IS NULL`, now, bound.String); err != nil {
return err
}
// The host must see the tombstone; an unbound claim has no host.
if err := bumpEpoch(tx); err != nil {
return err
}
}
return tx.Commit()
}
// HardDeleteVolume removes a tombstoned volume row once its host has reported
// the file gone, and the tombstoned claim that owned it. sql.ErrNoRows when the
// volume is absent, or when EITHER the volume or its claim is still live.
//
// The claim's tombstone is part of the guard, not an assumption: reaping the
// bytes of a claim somebody still holds would leave that claim Pending and
// re-bindable, so the next VM naming it would silently get an empty disk where
// the tenant expects their data. Deleting both rows together, or neither, is
// the only pair of outcomes that cannot lie.
func (s *Store) HardDeleteVolume(id string) error {
tx, err := s.db.Begin()
if err != nil {
return err
}
defer tx.Rollback()
var claimID string
if err := tx.QueryRow(`SELECT v.claim_id FROM volumes v
JOIN volume_claims c ON c.id = v.claim_id
WHERE v.id=? AND v.deleted_at IS NOT NULL AND c.deleted_at IS NOT NULL`, id).Scan(&claimID); err != nil {
return err
}
// Volume first: volumes.claim_id references the row deleted below.
if _, err := tx.Exec(`DELETE FROM volumes WHERE id=?`, id); err != nil {
return err
}
if _, err := tx.Exec(`DELETE FROM volume_claims WHERE id=?`, claimID); err != nil {
return err
}
if err := bumpEpoch(tx); err != nil {
return err
}
return tx.Commit()
}
// volumesOnHost lists the volumes living on one host and the claims they are
// placed for, in id order so a caller reporting them reports them the same way
// twice. It fully drains its result set before returning, so the caller may
// write inside the same transaction — the store's single connection allows
// nothing else.
func volumesOnHost(q querier, hostID string) (volumeIDs, claimIDs []string, err error) {
rows, err := q.Query(`SELECT id, claim_id FROM volumes WHERE host_id=? ORDER BY id`, hostID)
if err != nil {
return nil, nil, fmt.Errorf("list host volumes: %w", err)
}
defer rows.Close()
for rows.Next() {
var volID, claimID string
if err := rows.Scan(&volID, &claimID); err != nil {
return nil, nil, fmt.Errorf("scan host volume: %w", err)
}
volumeIDs = append(volumeIDs, volID)
claimIDs = append(claimIDs, claimID)
}
if err := rows.Err(); err != nil {
return nil, nil, err
}
return volumeIDs, claimIDs, nil
}
// ListVolumesForHost is the host's full volume set, tombstoned rows included:
// the snapshot carries both so the agent can reclaim.
func (s *Store) ListVolumesForHost(hostID string) ([]Volume, error) {
rows, err := s.db.Query(`SELECT id, host_id, claim_id, size_gb, created_at, deleted_at
FROM volumes WHERE host_id=? ORDER BY created_at, id`, hostID)
if err != nil {
return nil, err
}
defer rows.Close()
var out []Volume
for rows.Next() {
var v Volume
var createdAt string
var deletedAt sql.NullString
if err := rows.Scan(&v.ID, &v.HostID, &v.ClaimID, &v.SizeGB, &createdAt, &deletedAt); err != nil {
return nil, err
}
v.CreatedAt, _ = time.Parse(time.RFC3339, createdAt)
if deletedAt.Valid {
t, _ := time.Parse(time.RFC3339, deletedAt.String)
v.DeletedAt = &t
}
out = append(out, v)
}
return out, rows.Err()
}
// bindClaims runs inside CreateVM's transaction: every claim must be the VM's
// tenant's and live, unattached, and either Pending or already on this host.
// A Pending claim is bound HERE — a volumes row on vm.HostID — so binding is
// atomic with placement and never a background job.
func bindClaims(tx *sql.Tx, vm VM) error {
now := time.Now().UTC().Format(time.RFC3339)
for _, claimID := range vm.VolumeClaimIDs {
var sizeGB int64
var bound, boundHost, holder sql.NullString
err := tx.QueryRow(`SELECT c.size_gb, c.bound_volume_id, v.host_id, a.vm_id FROM volume_claims c
LEFT JOIN volumes v ON v.id=c.bound_volume_id
LEFT JOIN volume_attachments a ON a.claim_id=c.id
WHERE c.id=? AND c.tenant=? AND c.deleted_at IS NULL`, claimID, vm.Tenant).
Scan(&sizeGB, &bound, &boundHost, &holder)
switch {
case errors.Is(err, sql.ErrNoRows):
return ErrClaimNotFound
case err != nil:
return fmt.Errorf("lookup claim: %w", err)
case holder.Valid:
return &ClaimAttachedError{ClaimID: claimID, VMID: holder.String}
case bound.Valid && boundHost.String != vm.HostID:
return &ClaimPinnedError{ClaimID: claimID, HostID: boundHost.String}
}
if !bound.Valid {
volID := random.Hex(16)
if _, err := tx.Exec(`INSERT INTO volumes(id, host_id, claim_id, size_gb, created_at) VALUES (?,?,?,?,?)`,
volID, vm.HostID, claimID, sizeGB, now); err != nil {
return fmt.Errorf("place volume: %w", err)
}
if _, err := tx.Exec(`UPDATE volume_claims SET bound_volume_id=? WHERE id=?`, volID, claimID); err != nil {
return fmt.Errorf("bind claim: %w", err)
}
}
if _, err := tx.Exec(`INSERT INTO volume_attachments(claim_id, vm_id) VALUES (?,?)`, claimID, vm.ID); err != nil {
// The unique index is the race's referee: the other create won.
// The store runs one connection, so this is belt-and-braces.
if serr, ok := errors.AsType[*sqlite.Error](err); ok && serr.Code() == 2067 {
return &ClaimAttachedError{ClaimID: claimID, VMID: "another vm"}
}
return fmt.Errorf("attach claim: %w", err)
}
}
return nil
}
// volumeIDsForVM is the attachment order, which is the order the VM named its
// claims (rowid of the attachment row).
func volumeIDsForVM(q querier, vmID string) ([]string, error) {
rows, err := q.Query(`SELECT c.bound_volume_id FROM volume_attachments a
JOIN volume_claims c ON c.id=a.claim_id WHERE a.vm_id=? ORDER BY a.rowid`, vmID)
if err != nil {
return nil, err
}
defer rows.Close()
var out []string
for rows.Next() {
var id sql.NullString
if err := rows.Scan(&id); err != nil {
return nil, err
}
if id.Valid {
out = append(out, id.String)
}
}
return out, rows.Err()
}