package repository import ( "context" "encoding/json" "errors" "fmt" "log/slog" "os" "sync" "time" ) // journalState ist der Zustand einer Schreibsession. type journalState string const ( // journalStateOpen bedeutet: die Session schreibt noch. journalStateOpen journalState = "open" // journalStateCommitting bedeutet: der Abschluss läuft. // // Findet sich beim Start ein Journal in diesem Zustand, brach der Vorgang // mitten im Commit ab und muss geprüft werden. journalStateCommitting journalState = "committing" ) // sessionJournal hält den Fortschritt einer Schreibsession fest. // // Das Journal macht einen Abbruch nachvollziehbar: nach einem Absturz lässt // sich erkennen, welche Sessions offen blieben und aufgeräumt werden müssen // (PROMPT.md §82). type sessionJournal struct { // SessionID ist die Kennung der Schreibsession. SessionID string `json:"session_id"` // BackupID ist das Backup, das geschrieben wird. BackupID string `json:"backup_id"` // State ist der Zustand der Session. State journalState `json:"state"` // StartedAt ist der Beginn der Session in UTC. StartedAt time.Time `json:"started_at"` // UpdatedAt ist der Zeitpunkt der letzten Fortschreibung in UTC. UpdatedAt time.Time `json:"updated_at"` // Statistics sind die bisher erfassten Kennzahlen. Statistics SessionStatistics `json:"statistics"` // ProcessID ist der schreibende Prozess. ProcessID int `json:"process_id"` } // localWriter ist die Schreibsession eines lokalen Repositorys. type localWriter struct { // repository ist das Ziel-Repository. repository *LocalRepository // sessionID ist die Kennung dieser Session. sessionID string // backupID ist das zu schreibende Backup. backupID string // startedAt ist der Beginn der Session in UTC. startedAt time.Time // statisticsMutex schützt die Kennzahlen gegen gleichzeitige Schreiber. statisticsMutex sync.Mutex // statistics sind die laufenden Kennzahlen. statistics SessionStatistics // isClosed meldet, ob die Session bereits abgeschlossen oder abgebrochen ist. isClosed bool // logger protokolliert den Verlauf der Session. logger *slog.Logger } // BeginBackup öffnet eine Schreibsession (Schritt 1 des Commit-Protokolls). func (localRepository *LocalRepository) BeginBackup(beginContext context.Context, backupID string) (Writer, error) { if validationError := validateBackupIdentifier(backupID); validationError != nil { return nil, validationError } // Ein bereits vorhandenes Backup darf nicht überschrieben werden: das wäre // stiller Datenverlust (PROMPT.md §141). if _, statError := os.Stat(localRepository.manifestPath(backupID)); statError == nil { return nil, fmt.Errorf("zu der kennung %s existiert bereits ein backup", backupID) } // Ohne Schreibsperre könnten zwei Vorgänge denselben Katalog überschreiben. if localRepository.lockHandle == nil { return nil, fmt.Errorf("das repository wurde nur lesend geöffnet und kann nicht beschrieben werden") } sessionID := backupID startedAt := localRepository.timeSource().UTC() if directoryError := ensureDirectory(localRepository.stagingPath(sessionID)); directoryError != nil { return nil, directoryError } sessionWriter := &localWriter{ repository: localRepository, sessionID: sessionID, backupID: backupID, startedAt: startedAt, logger: localRepository.logger.With( slog.String("backup_id", backupID), slog.String("session_id", sessionID), ), } if journalError := sessionWriter.writeJournal(journalStateOpen); journalError != nil { return nil, journalError } sessionWriter.logger.Info("schreibsession geöffnet") return sessionWriter, nil } // WriteChunk legt einen Datenblock ab (Schritt 2 des Commit-Protokolls). func (sessionWriter *localWriter) WriteChunk(writeContext context.Context, chunkData []byte) (ChunkReference, bool, error) { sessionWriter.statisticsMutex.Lock() if sessionWriter.isClosed { sessionWriter.statisticsMutex.Unlock() return ChunkReference{}, false, ErrSessionClosed } sessionWriter.statisticsMutex.Unlock() // Ein abgebrochener Vorgang soll nicht weiterschreiben. if contextError := writeContext.Err(); contextError != nil { return ChunkReference{}, false, contextError } chunkIdentifier, wasNew, writeError := sessionWriter.repository.writeChunk(writeContext, chunkData) if writeError != nil { return ChunkReference{}, false, writeError } chunkLength := int64(len(chunkData)) sessionWriter.statisticsMutex.Lock() sessionWriter.statistics.LogicalBytes += chunkLength if wasNew { sessionWriter.statistics.ChunksWritten++ sessionWriter.statistics.StoredBytes += chunkLength } else { // Ein bereits vorhandener Chunk spart genau seine Länge ein. sessionWriter.statistics.ChunksDeduplicated++ sessionWriter.statistics.DeduplicatedBytes += chunkLength } sessionWriter.statisticsMutex.Unlock() return ChunkReference{ Identifier: chunkIdentifier, LogicalLength: chunkLength, StoredLength: chunkLength, }, wasNew, nil } // WriteTransformedChunk legt einen bereits umgewandelten Block ab und zählt ihn. // // Die Backup Engine komprimiert und verschlüsselt selbst; sie übergibt die // fertige Form. Gezählt wird beides: die Klartextlänge als verarbeitete // Datenmenge und die abgelegte Länge als tatsächlich geschriebene. // // Der Unterschied zwischen beiden ist der Gewinn aus Kompression — und der // Grund, warum die Klartextlänge als Parameter kommt: Aus dem Geheimtext lässt // sie sich nicht zurückrechnen. func (sessionWriter *localWriter) WriteTransformedChunk(writeContext context.Context, plaintextIdentifier string, storedData []byte, plaintextLength int64) (string, bool, error) { sessionWriter.statisticsMutex.Lock() if sessionWriter.isClosed { sessionWriter.statisticsMutex.Unlock() return "", false, ErrSessionClosed } sessionWriter.statisticsMutex.Unlock() if contextError := writeContext.Err(); contextError != nil { return "", false, contextError } storedDigest, wasNew, writeError := sessionWriter.repository.WriteTransformedChunk( writeContext, plaintextIdentifier, storedData) if writeError != nil { return "", false, writeError } storedLength := int64(len(storedData)) sessionWriter.statisticsMutex.Lock() sessionWriter.statistics.LogicalBytes += plaintextLength if wasNew { sessionWriter.statistics.ChunksWritten++ sessionWriter.statistics.StoredBytes += storedLength } else { // Ein bereits vorhandener Block spart genau seine abgelegte Länge. sessionWriter.statistics.ChunksDeduplicated++ sessionWriter.statistics.DeduplicatedBytes += storedLength } sessionWriter.statisticsMutex.Unlock() return storedDigest, wasNew, nil } // NoteDeduplicatedChunk vermerkt einen bereits vorhandenen Block. // // Gezaehlt wird die **Klartextlaenge**, nicht die abgelegte: Die Frage, die // diese Zahl beantwortet, lautet „wie viel musste nicht abgelegt werden" — und // die bemisst sich an den Ursprungsdaten. func (sessionWriter *localWriter) NoteDeduplicatedChunk(plaintextLength int64) { sessionWriter.statisticsMutex.Lock() defer sessionWriter.statisticsMutex.Unlock() if sessionWriter.isClosed { return } sessionWriter.statistics.LogicalBytes += plaintextLength sessionWriter.statistics.ChunksDeduplicated++ sessionWriter.statistics.DeduplicatedBytes += plaintextLength } // Commit schließt das Backup ab (Schritte 3 bis 7 des Commit-Protokolls). // // Der Ablauf folgt SYNCOVA_ARCHITECTURE.md §10: // // 3. Manifest schreiben // 4. Manifest prüfen // 5. Abschlussvermerk atomar setzen // 6. Katalog aktualisieren // 7. Erfolgszustand veröffentlichen // // Bricht der Vorgang vor Schritt 5 ab, bleibt kein sichtbares Backup zurück. func (sessionWriter *localWriter) Commit(commitContext context.Context, manifest *Manifest) error { sessionWriter.statisticsMutex.Lock() if sessionWriter.isClosed { sessionWriter.statisticsMutex.Unlock() return ErrSessionClosed } currentStatistics := sessionWriter.statistics sessionWriter.statisticsMutex.Unlock() if manifest == nil { return errors.New("ohne manifest kann kein backup abgeschlossen werden") } // Das Journal hält fest, dass der Abschluss begonnen hat. if journalError := sessionWriter.writeJournal(journalStateCommitting); journalError != nil { return journalError } // Die Angaben des Aufrufers werden mit den tatsächlich erfassten Werten // überschrieben: erfundene Kennzahlen wären eine Falschaussage (PROMPT.md §138). manifest.ManifestVersion = ManifestVersion manifest.BackupID = sessionWriter.backupID manifest.StartedAt = sessionWriter.startedAt manifest.CompletedAt = sessionWriter.repository.timeSource().UTC() manifest.Statistics = currentStatistics if manifest.ChainID == "" { // Ohne Kettenangabe bildet das Backup seine eigene Kette. manifest.ChainID = sessionWriter.backupID } if manifest.BackupType == "" { manifest.BackupType = BackupTypeFull } if manifest.ConsistencyLevel == "" { // Ohne ausdrückliche Zusicherung gilt die schwächste Stufe. Eine höhere // anzunehmen hiesse, eine nicht erbrachte Eigenschaft zu behaupten. manifest.ConsistencyLevel = ConsistencyCrash } // Jedes von dieser Programmversion geschriebene Manifest ist vollständig: // Auch eine Zusatzsicherung trägt die Blockverweise aller unveränderten // Objekte (Phase 6). Die Zusicherung wird hier festgehalten, damit die // Aufbewahrung ein Elternbackup löschen darf, ohne das Kind unbrauchbar zu // machen. manifest.SelfContainedRestore = true // Ein gehärtetes Repository setzt den Aufbewahrungsschutz selbst durch. // // Die Frist steht im Descriptor des Repositorys, nicht im Aufruf: Wer ein // Backup schreibt, soll die geltende Frist nicht unterlaufen können, indem // er einfach eine kürzere mitgibt. if sessionWriter.repository.descriptor.Immutable { earliestAllowedEnd := manifest.CompletedAt.Add(sessionWriter.repository.descriptor.RetentionPeriod()) if manifest.ImmutableUntil == nil || manifest.ImmutableUntil.Before(earliestAllowedEnd) { manifest.ImmutableUntil = &earliestAllowedEnd } } // Schritt 3 und 4: Prüfsumme bilden und sofort gegenprüfen. if sealError := sealManifest(manifest); sealError != nil { return sealError } if verifyError := VerifyManifest(manifest); verifyError != nil { return fmt.Errorf("das erzeugte manifest hat die eigene prüfung nicht bestanden: %w", verifyError) } encodedManifest, encodeError := encodeManifest(manifest) if encodeError != nil { return encodeError } // Schritt 5: Das atomare Umbenennen macht das Backup in einem Zug sichtbar. // Vorher existiert es für Leser nicht, nachher vollständig. manifestPermissions := dataFilePermissions if sessionWriter.repository.descriptor.Immutable { manifestPermissions = immutableFilePermissions } manifestFilePath := sessionWriter.repository.manifestPath(sessionWriter.backupID) if writeError := writeFileAtomically(manifestFilePath, encodedManifest, manifestPermissions); writeError != nil { return fmt.Errorf("das manifest konnte nicht abgelegt werden: %w", writeError) } // Der Löschschutz wird **nach** dem Sichtbarwerden gesetzt: Vorher liegt das // Manifest unter temporärem Namen, und ein unveränderliches Ziel liesse sich // nicht mehr per rename ersetzen — der Commit selbst schlüge fehl. // // Gelingt das Kennzeichen nicht, bleibt es beim Software-Schutz. Ein // geschriebenes Backup ohne technischen Löschschutz ist besser als gar // keines; verschwiegen wird es nicht, sondern über die gemessene // Durchsetzungsstufe ausgewiesen. if sessionWriter.repository.descriptor.Immutable { if protectError := protectManifestFile(manifestFilePath); protectError != nil { sessionWriter.logger.Warn("das manifest konnte nicht gegen löschung geschützt werden; "+ "es gilt nur der schutz durch diese software", slog.String("grund", protectError.Error())) } } // Schritt 6: Der Katalog folgt dem Manifest. Schlägt er fehl, ist das // Backup dennoch vollständig und wiederherstellbar - der Katalog lässt // sich jederzeit neu aufbauen. if catalogError := sessionWriter.repository.appendToCatalog(manifest); catalogError != nil { sessionWriter.logger.Error("der katalog konnte nicht aktualisiert werden; das backup ist dennoch vollständig", slog.String("error", catalogError.Error())) } sessionWriter.finish() // Schritt 7: Erst jetzt gilt das Backup als erfolgreich. sessionWriter.logger.Info("backup abgeschlossen", slog.Int64("chunks_written", currentStatistics.ChunksWritten), slog.Int64("chunks_deduplicated", currentStatistics.ChunksDeduplicated), slog.Int64("logical_bytes", currentStatistics.LogicalBytes), slog.Int64("stored_bytes", currentStatistics.StoredBytes), ) return nil } // Abort bricht die Session ab und räumt auf. // // Bereits geschriebene Chunks bleiben liegen: sie sind inhaltsadressiert und // werden von einem späteren Lauf wiederverwendet. Ein Manifest entsteht nicht, // weshalb kein unvollständiges Backup sichtbar wird. func (sessionWriter *localWriter) Abort(abortContext context.Context) error { sessionWriter.statisticsMutex.Lock() if sessionWriter.isClosed { sessionWriter.statisticsMutex.Unlock() return nil } sessionWriter.statisticsMutex.Unlock() sessionWriter.logger.Warn("schreibsession abgebrochen", slog.Int64("chunks_written", sessionWriter.statistics.ChunksWritten)) sessionWriter.finish() return nil } // Statistics liefert die bisher erfassten Kennzahlen. func (sessionWriter *localWriter) Statistics() SessionStatistics { sessionWriter.statisticsMutex.Lock() defer sessionWriter.statisticsMutex.Unlock() return sessionWriter.statistics } // finish räumt Journal und Arbeitsverzeichnis der Session weg. func (sessionWriter *localWriter) finish() { sessionWriter.statisticsMutex.Lock() sessionWriter.isClosed = true sessionWriter.statisticsMutex.Unlock() // Aufräumfehler werden protokolliert, aber nicht weitergereicht: das Backup // selbst ist zu diesem Zeitpunkt bereits abgeschlossen. if removeError := os.Remove(sessionWriter.repository.journalPath(sessionWriter.sessionID)); removeError != nil && !errors.Is(removeError, os.ErrNotExist) { sessionWriter.logger.Warn("das journal konnte nicht entfernt werden", slog.String("error", removeError.Error())) } if removeError := os.RemoveAll(sessionWriter.repository.stagingPath(sessionWriter.sessionID)); removeError != nil { sessionWriter.logger.Warn("das arbeitsverzeichnis konnte nicht entfernt werden", slog.String("error", removeError.Error())) } } // writeJournal schreibt den aktuellen Zustand der Session fort. func (sessionWriter *localWriter) writeJournal(currentState journalState) error { sessionWriter.statisticsMutex.Lock() currentStatistics := sessionWriter.statistics sessionWriter.statisticsMutex.Unlock() journalEntry := sessionJournal{ SessionID: sessionWriter.sessionID, BackupID: sessionWriter.backupID, State: currentState, StartedAt: sessionWriter.startedAt, UpdatedAt: sessionWriter.repository.timeSource().UTC(), Statistics: currentStatistics, ProcessID: os.Getpid(), } encodedJournal, marshalError := json.MarshalIndent(journalEntry, "", " ") if marshalError != nil { return fmt.Errorf("das journal konnte nicht erzeugt werden: %w", marshalError) } if writeError := writeFileAtomically(sessionWriter.repository.journalPath(sessionWriter.sessionID), append(encodedJournal, '\n'), dataFilePermissions); writeError != nil { return fmt.Errorf("das journal konnte nicht geschrieben werden: %w", writeError) } return nil } // StaleSession beschreibt eine nach einem Abbruch zurückgebliebene Session. type StaleSession struct { // SessionID ist die Kennung der Session. SessionID string `json:"session_id"` // BackupID ist das betroffene Backup. BackupID string `json:"backup_id"` // State ist der zuletzt vermerkte Zustand. State string `json:"state"` // StartedAt ist der Beginn der Session in UTC. StartedAt time.Time `json:"started_at"` // HasManifest meldet, ob trotz offener Session ein Manifest vorliegt. // // Trifft das zu, wurde der Abschluss vollzogen und nur das Aufräumen // unterbrochen — das Backup ist verwendbar. HasManifest bool `json:"has_manifest"` } // FindStaleSessions sucht Sessions, die ein früherer Lauf offen gelassen hat. // // Der Aufruf gehört an den Start des Dienstes: ein Abbruch darf nicht // stillschweigend Reste hinterlassen (PROMPT.md §140). func (localRepository *LocalRepository) FindStaleSessions(scanContext context.Context) ([]StaleSession, error) { journalDirectory := localRepository.rootPath + string(os.PathSeparator) + directoryJournals directoryEntries, readError := os.ReadDir(journalDirectory) if readError != nil { if errors.Is(readError, os.ErrNotExist) { return nil, nil } return nil, fmt.Errorf("das journalverzeichnis konnte nicht gelesen werden: %w", readError) } staleSessions := make([]StaleSession, 0) for _, directoryEntry := range directoryEntries { if directoryEntry.IsDir() || !hasSuffix(directoryEntry.Name(), journalExtension) { continue } rawJournal, journalReadError := os.ReadFile(journalDirectory + string(os.PathSeparator) + directoryEntry.Name()) if journalReadError != nil { continue } var journalEntry sessionJournal if unmarshalError := json.Unmarshal(rawJournal, &journalEntry); unmarshalError != nil { continue } _, manifestStatError := os.Stat(localRepository.manifestPath(journalEntry.BackupID)) staleSessions = append(staleSessions, StaleSession{ SessionID: journalEntry.SessionID, BackupID: journalEntry.BackupID, State: string(journalEntry.State), StartedAt: journalEntry.StartedAt, HasManifest: manifestStatError == nil, }) } return staleSessions, nil } // CleanupStaleSession räumt eine zurückgebliebene Session weg. // // Ein bereits geschriebenes Manifest bleibt unangetastet: der Abschluss war // vollzogen, nur das Aufräumen wurde unterbrochen. func (localRepository *LocalRepository) CleanupStaleSession(cleanupContext context.Context, sessionID string) error { if validationError := validateBackupIdentifier(sessionID); validationError != nil { return validationError } if removeError := os.Remove(localRepository.journalPath(sessionID)); removeError != nil && !errors.Is(removeError, os.ErrNotExist) { return fmt.Errorf("das journal konnte nicht entfernt werden: %w", removeError) } if removeError := os.RemoveAll(localRepository.stagingPath(sessionID)); removeError != nil { return fmt.Errorf("das arbeitsverzeichnis konnte nicht entfernt werden: %w", removeError) } localRepository.logger.Info("zurückgebliebene schreibsession aufgeräumt", slog.String("session_id", sessionID)) return nil }