()
| 305 | } |
| 306 | |
| 307 | func (n *Node) triggerSnapshotIfNeed() error { |
| 308 | if n.appliedIndex-n.snapshotIndex <= n.snapshotThreshold.Load() { |
| 309 | return nil |
| 310 | } |
| 311 | snapshotBytes, err := n.dataStore.GetDataStoreSnapshot() |
| 312 | if err != nil { |
| 313 | return err |
| 314 | } |
| 315 | snap, err := n.dataStore.raftStorage.CreateSnapshot(n.appliedIndex, &n.confState, snapshotBytes) |
| 316 | if err != nil { |
| 317 | return err |
| 318 | } |
| 319 | if err := n.dataStore.saveSnapshot(snap); err != nil { |
| 320 | return err |
| 321 | } |
| 322 | |
| 323 | compactIndex := uint64(1) |
| 324 | if n.appliedIndex > n.compactThreshold.Load() { |
| 325 | compactIndex = n.appliedIndex - n.compactThreshold.Load() |
| 326 | } |
| 327 | if err := n.dataStore.raftStorage.Compact(compactIndex); err != nil && !errors.Is(err, raft.ErrCompacted) { |
| 328 | return err |
| 329 | } |
| 330 | n.snapshotIndex = n.appliedIndex |
| 331 | return nil |
| 332 | } |
| 333 | |
| 334 | func (n *Node) Set(ctx context.Context, key string, value []byte) error { |
| 335 | bytes, err := json.Marshal(&Event{ |
no test coverage detected