checkpoint.go

package qi

import (
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"

	"github.com/iamseth/yin"
)

// CheckpointCommitter publishes complete qi checkpoint bytes to an
// application-owned durable authority.
//
// CommitCheckpoint must return (true, nil) after successful publication. An
// error before publication must return published false. If publication occurred
// but its durability cannot be confirmed, it must return published true with a
// non-nil error. Returning false with a nil error is invalid.
//
// The checkpoint slice is owned by the caller: CommitCheckpoint must not mutate
// it or retain it after returning. The application must give a Store exclusive
// authority over the corresponding checkpoint while that Store uses the
// committer. Calls are serialized by Store; implementations need not support
// concurrent or reentrant calls.
//
// Checkpoint bytes are the complete qi checkpoint envelope, including document
// identity and a yin snapshot. They are not visible JSON, a bare yin snapshot,
// or a delta.
type CheckpointCommitter interface {
	CommitCheckpoint(checkpoint []byte) (published bool, err error)
}

// ErrDurabilityUncertain means a checkpoint was published, but the committer
// could not confirm its durability. The Store adopts the published checkpoint
// before returning an error wrapping this sentinel.
var ErrDurabilityUncertain = errors.New("qi: checkpoint durability uncertain")

const checkpointValidationReplica yin.ReplicaID = "qi-checkpoint-validation"

func encodeCheckpoint(artifact SnapshotArtifact) ([]byte, error) {
	data, err := json.Marshal(artifact)
	if err != nil {
		return nil, err
	}
	if _, _, err := decodeCheckpoint(checkpointValidationReplica, data); err != nil {
		return nil, err
	}
	return data, nil
}

func decodeCheckpoint(replica yin.ReplicaID, data []byte) (*JSONFile, SnapshotArtifact, error) {
	return decodeCheckpointForDocument(replica, "", data)
}

// decodeCheckpointForDocument checks the envelope and expected document before
// handing its payload to yin. An empty expectedDocumentID accepts any document.
func decodeCheckpointForDocument(replica yin.ReplicaID, expectedDocumentID string, data []byte) (*JSONFile, SnapshotArtifact, error) {
	var artifact SnapshotArtifact
	if err := json.Unmarshal(data, &artifact); err != nil {
		return nil, SnapshotArtifact{}, fmt.Errorf("decode qi checkpoint: %w", err)
	}
	if err := validateSnapshotArtifact(artifact); err != nil {
		return nil, SnapshotArtifact{}, err
	}
	if expectedDocumentID != "" && artifact.Identity.DocumentID != expectedDocumentID {
		return nil, SnapshotArtifact{}, fmt.Errorf("%w: expected %q, checkpoint has %q", ErrDocumentMismatch, expectedDocumentID, artifact.Identity.DocumentID)
	}
	file, err := ParseJSONFileSnapshot(replica, artifact.Payload)
	if err != nil {
		return nil, SnapshotArtifact{}, fmt.Errorf("%w: yin snapshot: %v", ErrMalformedSyncArtifact, err)
	}
	return file, artifact, nil
}

type fileCheckpointCommitter struct {
	path          string
	perm          os.FileMode
	rename        func(string, string) error
	syncDirectory func(*os.File) error
}

func newFileCheckpointCommitter(path string, perm os.FileMode) *fileCheckpointCommitter {
	return &fileCheckpointCommitter{path: path, perm: perm}
}

func (c *fileCheckpointCommitter) CommitCheckpoint(checkpoint []byte) (bool, error) {
	if c.path == "" {
		return false, errors.New("qi: snapshot path is empty")
	}
	dir := filepath.Dir(c.path)
	if dir != "." && dir != "" {
		if err := os.MkdirAll(dir, 0o755); err != nil {
			return false, err
		}
	}

	dirFile, err := os.Open(dir)
	if err != nil {
		return false, fmt.Errorf("open snapshot directory: %w", err)
	}
	defer dirFile.Close()

	tmp, err := os.CreateTemp(dir, "."+filepath.Base(c.path)+"-*")
	if err != nil {
		return false, err
	}
	tmpPath := tmp.Name()
	cleanup := true
	defer func() {
		if cleanup {
			_ = os.Remove(tmpPath)
		}
	}()

	if _, err := tmp.Write(checkpoint); err != nil {
		_ = tmp.Close()
		return false, err
	}
	if err := tmp.Chmod(c.perm); err != nil {
		_ = tmp.Close()
		return false, err
	}
	if err := tmp.Sync(); err != nil {
		_ = tmp.Close()
		return false, err
	}
	if err := tmp.Close(); err != nil {
		return false, err
	}
	rename := c.rename
	if rename == nil {
		rename = os.Rename
	}
	if err := rename(tmpPath, c.path); err != nil {
		return false, err
	}
	cleanup = false

	syncDirectory := c.syncDirectory
	if syncDirectory == nil {
		syncDirectory = func(file *os.File) error { return file.Sync() }
	}
	if err := syncDirectory(dirFile); err != nil {
		return true, fmt.Errorf("sync snapshot directory: %w", err)
	}
	return true, nil
}