package backupengine import ( "context" "crypto/sha256" "encoding/hex" "errors" "fmt" "io" "log/slog" "time" "github.com/syncova/syncova/packages/platform/crypto" "github.com/syncova/syncova/packages/platform/logging" "github.com/syncova/syncova/packages/platform/ratelimit" "github.com/syncova/syncova/packages/repository" ) // Engine führt Backups und Wiederherstellungen aus. type Engine struct { // targetRepository ist das Ziel-Repository. targetRepository *repository.LocalRepository // secretStore verschlüsselt die Datenschlüssel der Backups. secretStore crypto.SecretStore // logger protokolliert den Verlauf. logger *slog.Logger } // NewEngine erzeugt eine Backup Engine. // // Der Secret Store darf nil sein; dann sind ausschließlich unverschlüsselte // Backups möglich. Das ist nur für Testzwecke gedacht: PROMPT.md §119 verlangt // Verschlüsselung als Standard. func NewEngine(targetRepository *repository.LocalRepository, secretStore crypto.SecretStore, baseLogger *slog.Logger) *Engine { return &Engine{ targetRepository: targetRepository, secretStore: secretStore, logger: logging.WithComponent(baseLogger, "backup-engine"), } } // Repository liefert das Ziel-Repository. // // Der Zugriff wird gebraucht, um vor einer Zusatzsicherung das Manifest des // Elternbackups zu lesen. Die Engine selbst trifft diese Entscheidung nicht — // welches Backup das Elternbackup ist, hängt von der Quelle ab und damit von // einer Schicht über ihr. func (engine *Engine) Repository() *repository.LocalRepository { return engine.targetRepository } // BackupOptions steuern einen Backup-Lauf. type BackupOptions struct { // BackupID ist die Kennung des zu erzeugenden Backups. BackupID string // ChainID verbindet das Backup mit seiner Kette. ChainID string // ParentBackupID benennt das Elternbackup einer Zusatzsicherung. ParentBackupID string // BackupType ist die Art des Backups. BackupType repository.BackupType // Source beschreibt die Herkunft der Daten. Source repository.SourceInformation // CompressionLevel ist die gewünschte Kompressionsstufe. CompressionLevel CompressionLevel // EncryptionEnabled schaltet die Verschlüsselung ein. // // Sie ist der Standard; ein Abschalten muss ausdrücklich geschehen // (PROMPT.md §12, §120). EncryptionEnabled bool // WorkerCount ist die Zahl paralleler Arbeiter; 0 wählt einen sinnvollen Wert. WorkerCount int // QueueDepth ist die Tiefe der Warteschlangen; 0 wählt den Standardwert. QueueDepth int // ChunkerOptions steuern die Blockfindung. ChunkerOptions ChunkerOptions // ProgressCallback meldet den Fortschritt. ProgressCallback ProgressCallback // CreatedByVersion ist die erzeugende Programmversion. CreatedByVersion string // BandwidthLimiter begrenzt den Lesedurchsatz von den Quellen. // // Er gilt für den **gesamten** Lauf und wird über alle Quellen geteilt. // Je Quelle einen eigenen zu führen ergäbe bei einem Auftrag mit drei // Verzeichnissen das Dreifache der vereinbarten Rate. // // Begrenzt wird das Lesen, nicht das Schreiben. Weil die Pipeline mit // Gegendruck arbeitet, bremst das den gesamten Ablauf: Was nicht gelesen // wird, wird auch nicht gehasht, komprimiert, verschlüsselt und abgelegt. // Ein Begrenzer an dieser einen Stelle bindet damit die gesamte Last. // // Ein nil-Begrenzer bedeutet unbegrenzt. BandwidthLimiter *ratelimit.Limiter } // BackupSource beschreibt ein zu sicherndes Objekt. type BackupSource struct { // Path ist der Pfad des Objekts in der Quelle. Path string // EntryType benennt die Art des Objekts. EntryType string // SizeBytes ist die erwartete Größe; 0 bedeutet unbekannt. SizeBytes int64 // ModifiedAt ist der Änderungszeitpunkt. ModifiedAt time.Time // Mode sind die Dateirechte in oktaler Schreibweise. // // Ohne sie wäre eine Wiederherstellung unvollständig: eine zurückgespielte // Datei trüge die Standardrechte statt ihrer ursprünglichen. Mode string // LinkTarget ist das Ziel eines symbolischen Verweises. LinkTarget string // Reader liefert den Inhalt. // // Er bleibt nil bei Objekten ohne Daten — Verzeichnisse und symbolische // Verweise tragen nur Metadaten. Reader io.Reader // ReusedChunks übernimmt die Blockverweise eines früheren Backups. // // Damit sichert eine Zusatzsicherung ein unverändertes Objekt, ohne es zu // lesen. Die Ersparnis liegt in Lese- und Rechenzeit: der Speicherplatz // wäre ohnehin durch die Deduplizierung eingespart worden. // // Ob ein Objekt unverändert ist, entscheidet der Aufrufer. Die Engine // kennt keine Dateisysteme und darf diese Frage nicht beantworten — sie // prüft aber, dass jeder übernommene Block wirklich im Repository liegt. ReusedChunks []repository.ChunkReference // ContentHash ist die Inhaltsprüfsumme eines übernommenen Objekts. ContentHash string } // BackupResult beschreibt einen abgeschlossenen Backup-Lauf. type BackupResult struct { // BackupID ist die Kennung des erzeugten Backups. BackupID string `json:"backup_id"` // Progress sind die Endkennzahlen. Progress Progress `json:"progress"` // EntryCount ist die Zahl gesicherter Objekte. EntryCount int `json:"entry_count"` // Encrypted meldet, ob verschlüsselt wurde. Encrypted bool `json:"encrypted"` // CompressionAlgorithm benennt das verwendete Kompressionsverfahren. CompressionAlgorithm string `json:"compression_algorithm,omitempty"` // Duration ist die Gesamtdauer. Duration time.Duration `json:"duration"` } // ErrEncryptionUnavailable meldet eine angeforderte, aber nicht mögliche Verschlüsselung. var ErrEncryptionUnavailable = errors.New("die verschlüsselung wurde angefordert, es ist aber kein schlüsselspeicher eingerichtet") // Backup sichert die übergebenen Quellen in das Repository. // // Der Ablauf entspricht Phase 4 des Implementierungsplans: Lesen, Chunking, // Hash, Deduplizierung, Kompression, Verschlüsselung, Schreiben, Manifest, // Commit. Kein Schritt hält mehr als wenige Blöcke gleichzeitig im Speicher. func (engine *Engine) Backup(backupContext context.Context, backupOptions BackupOptions, backupSources []BackupSource) (BackupResult, error) { startTime := time.Now() if len(backupSources) == 0 { // Ein Backup ohne Quellen wäre ein leeres Gebilde, das sich als // erfolgreiche Sicherung ausgäbe (PROMPT.md §138). return BackupResult{}, errors.New("es wurde keine quelle zum sichern angegeben") } // Der Datenschlüssel gehört zum Repository, nicht zum einzelnen Backup. // // Ein Schlüssel je Backup machte jede Deduplizierung über Backupgrenzen // hinweg unmöglich: ein späteres Backup verwiese auf Blöcke, die mit einem // fremden Schlüssel verschlüsselt und damit für es unlesbar wären. var dataEncryptionKey []byte var keyVersion string if backupOptions.EncryptionEnabled { if engine.secretStore == nil { return BackupResult{}, ErrEncryptionUnavailable } repositoryKey, keyError := engine.targetRepository.LoadOrCreateDataKey(backupContext, engine.secretStore) if keyError != nil { return BackupResult{}, keyError } storedKeyVersion, versionError := engine.targetRepository.DataKeyVersion() if versionError != nil { return BackupResult{}, fmt.Errorf("die schlüsselversion konnte nicht ermittelt werden: %w", versionError) } dataEncryptionKey = repositoryKey keyVersion = storedKeyVersion } chunkTransformer, transformerError := NewChunkTransformer(TransformerOptions{ CompressionLevel: backupOptions.CompressionLevel, DataEncryptionKey: dataEncryptionKey, }) if transformerError != nil { return BackupResult{}, transformerError } defer chunkTransformer.Close() workerCount, workerError := validateWorkerCount(backupOptions.WorkerCount) if workerError != nil { return BackupResult{}, workerError } queueDepth := backupOptions.QueueDepth if queueDepth <= 0 { queueDepth = defaultQueueDepth } progressReporter := NewProgressReporter(backupOptions.ProgressCallback, 0) // Die Schreibsession entsteht **vor** der Pipeline: Diese schreibt ueber // sie, damit die Kennzahlen des Laufs mitgezaehlt werden und ins Manifest // gelangen. backupWriter, beginError := engine.targetRepository.BeginBackup(backupContext, backupOptions.BackupID) if beginError != nil { return BackupResult{}, beginError } processingPipeline := &pipeline{ transformer: chunkTransformer, sink: &repositorySink{ backupWriter: backupWriter, targetRepository: engine.targetRepository, }, workerCount: workerCount, queueDepth: queueDepth, progressReporter: progressReporter, } engine.logger.Info("backup gestartet", slog.String("backup_id", backupOptions.BackupID), slog.Int("quellen", len(backupSources)), slog.Int("arbeiter", workerCount), slog.String("kompression", string(backupOptions.CompressionLevel)), slog.Bool("verschluesselt", backupOptions.EncryptionEnabled), slog.String("bandbreite", ratelimit.FormatBandwidthLimit(backupOptions.BandwidthLimiter.BytesPerSecond()))) manifestEntries := make([]repository.ManifestEntry, 0, len(backupSources)) for _, backupSource := range backupSources { if contextError := backupContext.Err(); contextError != nil { // Ein Abbruch lässt kein sichtbares Backup zurück. _ = backupWriter.Abort(backupContext) return BackupResult{}, contextError } // Ein unverändertes Objekt wird nicht erneut gelesen; seine // Blockverweise stammen aus dem Elternbackup. // // Das neue Manifest bleibt dadurch vollständig: es beschreibt den // gesamten Bestand, nicht nur die Änderungen. Eine Wiederherstellung // braucht deshalb nie die Kette — sie liest ein einziges Manifest. // Das ist der Grund, warum das Löschen eines alten Backups ein // neueres nicht beschädigen kann. if len(backupSource.ReusedChunks) > 0 { manifestEntry, reuseError := engine.reuseChunks(backupContext, backupSource, progressReporter) if reuseError != nil { _ = backupWriter.Abort(backupContext) return BackupResult{}, reuseError } manifestEntries = append(manifestEntries, manifestEntry) continue } // Objekte ohne Datenstrom durchlaufen die Pipeline nicht: ein // Verzeichnis oder ein symbolischer Verweis trägt nur Metadaten. // Sie gehören dennoch ins Manifest, sonst ginge bei der // Wiederherstellung die Struktur samt Rechten verloren. if backupSource.Reader == nil { manifestEntries = append(manifestEntries, repository.ManifestEntry{ Path: backupSource.Path, EntryType: backupSource.EntryType, ModifiedAt: backupSource.ModifiedAt, Mode: backupSource.Mode, LinkTarget: backupSource.LinkTarget, }) continue } // Die Bandbreitengrenze wirkt vor allem anderen: Der Begrenzer umhüllt // den Quelldatenstrom, bevor Prüfsumme und Chunking daran arbeiten. // Ohne Grenze wird der Datenstrom unverändert durchgereicht — eine // Hülle, die nichts tut, kostet bei jedem Block einen Aufruf mehr. limitedSourceReader := ratelimit.NewLimitedReader(backupContext, backupSource.Reader, backupOptions.BandwidthLimiter) // Der Inhaltshash entsteht nebenher, damit sich die Wiederherstellung // eines einzelnen Objekts prüfen lässt, ohne alle Blöcke einzeln // nachzurechnen. contentDigest := sha256.New() countingReader := io.TeeReader(limitedSourceReader, contentDigest) // Die bekannte Quellgroesse geht mit: Ohne sie legt der Chunker den // Lesepuffer in Hoechstblockgroesse an — vier Megabyte, auch fuer eine // Datei von sechzehn Kilobyte. sourceChunkerOptions := backupOptions.ChunkerOptions sourceChunkerOptions.ExpectedSize = backupSource.SizeBytes chunkReferences, processError := processingPipeline.process(backupContext, countingReader, sourceChunkerOptions) if processError != nil { _ = backupWriter.Abort(backupContext) return BackupResult{}, fmt.Errorf("die quelle %q konnte nicht gesichert werden: %w", backupSource.Path, processError) } var totalLogicalBytes int64 for _, chunkReference := range chunkReferences { totalLogicalBytes += chunkReference.LogicalLength } manifestEntries = append(manifestEntries, repository.ManifestEntry{ Path: backupSource.Path, EntryType: backupSource.EntryType, SizeBytes: totalLogicalBytes, ModifiedAt: backupSource.ModifiedAt, Mode: backupSource.Mode, Chunks: chunkReferences, ContentHash: hex.EncodeToString(contentDigest.Sum(nil)), }) } compressionName, encryptionName := chunkTransformer.AlgorithmNames() backupManifest := &repository.Manifest{ ChainID: backupOptions.ChainID, ParentBackupID: backupOptions.ParentBackupID, BackupType: backupOptions.BackupType, Source: backupOptions.Source, Entries: manifestEntries, CompressionAlgorithm: compressionName, EncryptionKeyVersion: keyVersion, CreatedByVersion: backupOptions.CreatedByVersion, } // Das Manifest vermerkt lediglich, dass verschlüsselt wurde. Der Schlüssel // selbst liegt im Repository - er gilt für alle seine Backups. if encryptionName != "" { backupManifest.Source.Attributes = mergeAttributes(backupManifest.Source.Attributes, map[string]string{ encryptionAlgorithmAttribute: encryptionName, }) } if commitError := backupWriter.Commit(backupContext, backupManifest); commitError != nil { return BackupResult{}, commitError } finalProgress := progressReporter.ReportFinal() engine.logger.Info("backup abgeschlossen", slog.String("backup_id", backupOptions.BackupID), slog.Int64("bytes_verarbeitet", finalProgress.BytesProcessed), slog.Int64("bytes_abgelegt", finalProgress.BytesWritten), slog.Int64("chunks_neu", finalProgress.ChunksWritten), slog.Int64("chunks_dedupliziert", finalProgress.ChunksDeduplicated), slog.String("dauer", time.Since(startTime).String())) return BackupResult{ BackupID: backupOptions.BackupID, Progress: finalProgress, EntryCount: len(manifestEntries), Encrypted: backupOptions.EncryptionEnabled, CompressionAlgorithm: compressionName, Duration: time.Since(startTime), }, nil } // ErrReusedChunkMissing meldet einen übernommenen Block, der im Repository fehlt. var ErrReusedChunkMissing = errors.New("ein aus dem elternbackup übernommener block fehlt im repository") // reuseChunks übernimmt die Blockverweise eines unveränderten Objekts. // // Vor der Übernahme wird die Existenz jedes Blocks geprüft. Ohne diese Prüfung // entstünde ein Manifest, das sich als vollständiges Backup ausgibt, während // seine Daten fehlen — genau die Art von stillem Fehler, die PROMPT.md §138 // verbietet. Geprüft wird nur die Existenz, nicht der Inhalt: den Inhalt prüft // der Integritätslauf, und ihn hier zu lesen hübe den Zweck der Zusatzsicherung // auf. func (engine *Engine) reuseChunks(reuseContext context.Context, backupSource BackupSource, progressReporter *ProgressReporter) (repository.ManifestEntry, error) { var totalLogicalBytes int64 for _, chunkReference := range backupSource.ReusedChunks { chunkExists, existenceError := engine.targetRepository.HasChunk(reuseContext, chunkReference.Identifier) if existenceError != nil { return repository.ManifestEntry{}, fmt.Errorf("der block %s des objekts %q konnte nicht geprüft werden: %w", chunkReference.Identifier, backupSource.Path, existenceError) } if !chunkExists { return repository.ManifestEntry{}, fmt.Errorf("%w (objekt %q, block %s)", ErrReusedChunkMissing, backupSource.Path, chunkReference.Identifier) } totalLogicalBytes += chunkReference.LogicalLength progressReporter.recordReusedChunk(chunkReference.LogicalLength) } return repository.ManifestEntry{ Path: backupSource.Path, EntryType: backupSource.EntryType, SizeBytes: totalLogicalBytes, ModifiedAt: backupSource.ModifiedAt, Mode: backupSource.Mode, LinkTarget: backupSource.LinkTarget, Chunks: backupSource.ReusedChunks, ContentHash: backupSource.ContentHash, }, nil } // encryptionAlgorithmAttribute benennt das Verschlüsselungsverfahren im Manifest. // // Der Datenschlüssel selbst steht nicht hier, sondern im Repository: er gilt // für alle seine Backups und ermöglicht damit die Deduplizierung. const encryptionAlgorithmAttribute = "syncova.encryption_algorithm" // mergeAttributes fügt Attribute zusammen, ohne die Vorlage zu verändern. func mergeAttributes(existingAttributes map[string]string, additionalAttributes map[string]string) map[string]string { mergedAttributes := make(map[string]string, len(existingAttributes)+len(additionalAttributes)) for attributeName, attributeValue := range existingAttributes { mergedAttributes[attributeName] = attributeValue } for attributeName, attributeValue := range additionalAttributes { mergedAttributes[attributeName] = attributeValue } return mergedAttributes }