diff options
Diffstat (limited to 'state')
| -rw-r--r-- | state/memory_backend.go | 5 | ||||
| -rw-r--r-- | state/pebble_backend.go | 41 | ||||
| -rw-r--r-- | state/pebble_state.go | 1 |
3 files changed, 47 insertions, 0 deletions
diff --git a/state/memory_backend.go b/state/memory_backend.go index cce8194..de26cab 100644 --- a/state/memory_backend.go +++ b/state/memory_backend.go | |||
| @@ -58,3 +58,8 @@ func (ms *MemoryStateBackend) SetCurrentNamespace(namespace any) error { | |||
| 58 | ms.currentNamespace = namespace | 58 | ms.currentNamespace = namespace |
| 59 | return nil | 59 | return nil |
| 60 | } | 60 | } |
| 61 | |||
| 62 | func (ms *MemoryStateBackend) Snapshot(checkpointID uint64, targetDir string) error { | ||
| 63 | // TODO: temporary return nil to avoid breaking | ||
| 64 | return nil | ||
| 65 | } | ||
diff --git a/state/pebble_backend.go b/state/pebble_backend.go index b510f8f..f1aae0c 100644 --- a/state/pebble_backend.go +++ b/state/pebble_backend.go | |||
| @@ -4,6 +4,8 @@ import ( | |||
| 4 | "context" | 4 | "context" |
| 5 | "encoding/binary" | 5 | "encoding/binary" |
| 6 | "fmt" | 6 | "fmt" |
| 7 | "os" | ||
| 8 | "path/filepath" | ||
| 7 | "sync" | 9 | "sync" |
| 8 | 10 | ||
| 9 | "flux/core" | 11 | "flux/core" |
| @@ -133,4 +135,43 @@ func (p *PebbleStateBackend) Close() error { | |||
| 133 | return p.db.Close() | 135 | return p.db.Close() |
| 134 | } | 136 | } |
| 135 | 137 | ||
| 138 | // Snapshot atomic | ||
| 139 | func (p *PebbleStateBackend) Snapshot(checkpointID uint64, targetDir string) error { | ||
| 140 | // get parent dir | ||
| 141 | parentDir := filepath.Dir(targetDir) | ||
| 136 | 142 | ||
| 143 | if err := os.MkdirAll(parentDir, 0o755); err != nil { | ||
| 144 | return err | ||
| 145 | } | ||
| 146 | |||
| 147 | // create temp dir | ||
| 148 | tmpDir := targetDir + ".tmp" | ||
| 149 | _ = os.RemoveAll(tmpDir) | ||
| 150 | |||
| 151 | // rollback | ||
| 152 | success := false | ||
| 153 | defer func() { | ||
| 154 | if !success { | ||
| 155 | _ = os.RemoveAll(tmpDir) | ||
| 156 | } | ||
| 157 | }() | ||
| 158 | |||
| 159 | if err := p.db.Checkpoint(tmpDir); err != nil { | ||
| 160 | return fmt.Errorf("pebble checkpoint failed for ID %d: %w", checkpointID, err) | ||
| 161 | } | ||
| 162 | |||
| 163 | _ = os.RemoveAll(targetDir) | ||
| 164 | |||
| 165 | if err := os.Rename(tmpDir, targetDir); err != nil { | ||
| 166 | return fmt.Errorf("failed to commit checkpoint atomically from %s to %s: %w", tmpDir, targetDir, err) | ||
| 167 | } | ||
| 168 | |||
| 169 | // fsync the parent dir | ||
| 170 | if parentFile, err := os.Open(parentDir); err == nil { | ||
| 171 | _ = parentFile.Sync() | ||
| 172 | _ = parentFile.Close() | ||
| 173 | } | ||
| 174 | |||
| 175 | success = true | ||
| 176 | return nil | ||
| 177 | } | ||
diff --git a/state/pebble_state.go b/state/pebble_state.go index 0b60deb..1560ae8 100644 --- a/state/pebble_state.go +++ b/state/pebble_state.go | |||
| @@ -52,3 +52,4 @@ func (p *pebbleStateValue) Clear(ctx context.Context) error { | |||
| 52 | 52 | ||
| 53 | return p.backend.db.Delete(key, pebble.NoSync) | 53 | return p.backend.db.Delete(key, pebble.NoSync) |
| 54 | } | 54 | } |
| 55 | |||
