diff --git a/abciapp/app.go b/abciapp/app.go index 27f4c1f..87b2183 100644 --- a/abciapp/app.go +++ b/abciapp/app.go @@ -17,6 +17,7 @@ import ( "github.com/gbl08ma/stacktrace" "github.com/klauspost/compress/zstd" "github.com/samber/lo" + "tangled.org/gbl08ma.com/didplcbft/config" "tangled.org/gbl08ma.com/didplcbft/dbmtoiavldb" "tangled.org/gbl08ma.com/didplcbft/dbmtoiavldb/zstddict" "tangled.org/gbl08ma.com/didplcbft/plc" @@ -56,10 +57,15 @@ type DIDPLCApplication struct { blockChallengeCoordinator *blockChallengeCoordinator rangeChallengeCoordinator *rangeChallengeCoordinator + + // for snapshot creation: + plcConfig *config.PLCConfig + snapshotManagerLatestHeightChan chan int64 + lastSnapshotHeight int64 } // store and plc must be able to share transaction objects -func NewDIDPLCApplication(appContext context.Context, logger cmtlog.Logger, pv *privval.FilePV, treeDB dbm.DB, indexDB transaction.ExtendedDB, clearData func(), snapshotDirectory, stateSyncTempDir, didBloomFilterPath string, mempoolSubmitter types.MempoolSubmitter, blockHeaderGetter store.BlockHeaderGetter) (*DIDPLCApplication, *transaction.Factory, plc.PLC, func(), error) { +func NewDIDPLCApplication(appContext context.Context, logger cmtlog.Logger, pv *privval.FilePV, treeDB dbm.DB, indexDB transaction.ExtendedDB, clearData func(), snapshotDirectory, stateSyncTempDir, didBloomFilterPath string, mempoolSubmitter types.MempoolSubmitter, blockHeaderGetter store.BlockHeaderGetter, plcConfig *config.PLCConfig) (*DIDPLCApplication, *transaction.Factory, plc.PLC, func(), error) { mkTree := func() *iavl.MutableTree { // Using SpeedDefault appears to cause the processing time for ExecuteOperation to double on average // Using SpeedBetterCompression appears to cause the processing time to double again @@ -91,15 +97,17 @@ func NewDIDPLCApplication(appContext context.Context, logger cmtlog.Logger, pv * runnerContext, cancelRunnerContext := context.WithCancel(appContext) d := &DIDPLCApplication{ - runnerContext: runnerContext, - logger: logger.With("module", "plcapp"), - tree: tree, - indexDB: indexDB, - mempoolSubmitter: mempoolSubmitter, - snapshotDirectory: snapshotDirectory, - stateSyncTempDir: stateSyncTempDir, - blockHeaderGetter: blockHeaderGetter, - triggerBlockCreation: func() {}, + runnerContext: runnerContext, + logger: logger.With("module", "plcapp"), + tree: tree, + indexDB: indexDB, + mempoolSubmitter: mempoolSubmitter, + snapshotDirectory: snapshotDirectory, + stateSyncTempDir: stateSyncTempDir, + blockHeaderGetter: blockHeaderGetter, + triggerBlockCreation: func() {}, + plcConfig: plcConfig, + snapshotManagerLatestHeightChan: make(chan int64, 1), } if pv != nil { @@ -159,56 +167,17 @@ func NewDIDPLCApplication(appContext context.Context, logger cmtlog.Logger, pv * } }) - /*lastSnapshotVersion := tree.Version() - wg.Go(func() { - for { - select { - case <-closeCh: - return - case <-time.After(5 * time.Minute): - } - - treeVersion := tree.Version() - if treeVersion > int64(lastSnapshotVersion+10000) { - err = d.createSnapshot(treeVersion, filepath.Join(snapshotDirectory, "snapshot.tmp")) - if err != nil { - fmt.Println("FAILED TO TAKE SNAPSHOT", stacktrace.Propagate(err)) - } - fmt.Println("TOOK SNAPSHOT OF VERSION", treeVersion) - lastSnapshotVersion = treeVersion - } - } - - })*/ - - /*err = d.createSnapshot(tree.Version(), filepath.Join(snapshotDirectory, "snapshot.tmp")) - if err != nil { - return nil, nil, func() {}, stacktrace.Propagate(err) - }*/ - - /* - tree2 := iavl.NewMutableTree(dbm.NewMemDB(), 2048, false, iavl.NewNopLogger()) - importer, err := tree2.Import(tree.Version()) - if err != nil { - return nil, nil, func() {}, stacktrace.Propagate(err) - } - cimporter := iavl.NewCompressImporter(importer) - - st = time.Now() - for _, node := range nodes { - err := cimporter.Add(&node) - if err != nil { - return nil, nil, func() {}, stacktrace.Propagate(err) - } - } - err = importer.Commit() + if plcConfig != nil && plcConfig.SnapshotInterval > 0 && snapshotDirectory != "" { + h, err := store.Snapshot.MostRecentSnapshotHeight(snapshotDirectory) if err != nil { - return nil, nil, func() {}, stacktrace.Propagate(err) + return nil, nil, nil, cancelRunnerContext, stacktrace.Propagate(err) } + d.lastSnapshotHeight = int64(h) - fmt.Println("Took", time.Since(st), "to import", len(nodes), "nodes") - fmt.Println("Imported tree hash", hex.EncodeToString(tree2.Hash()), "and version", tree2.Version()) - */ + wg.Go(func() { + d.runSnapshotManager(runnerContext, snapshotDirectory) + }) + } return d, d.txFactory, d.plc, func() { cancelRunnerContext() diff --git a/abciapp/app_test.go b/abciapp/app_test.go index b347814..d7d9d1c 100644 --- a/abciapp/app_test.go +++ b/abciapp/app_test.go @@ -11,6 +11,7 @@ import ( cbornode "github.com/ipfs/go-ipld-cbor" "github.com/stretchr/testify/require" "tangled.org/gbl08ma.com/didplcbft/abciapp" + "tangled.org/gbl08ma.com/didplcbft/config" ) func txJSONToCBOR(t *testing.T, jsonBytes []byte) []byte { @@ -24,7 +25,7 @@ func txJSONToCBOR(t *testing.T, jsonBytes []byte) []byte { func TestCheckTx(t *testing.T) { logger := cmtlog.NewNopLogger() - app, _, _, cleanup, err := abciapp.NewDIDPLCApplication(t.Context(), logger, nil, dbm.NewMemDB(), memDBWrapper{dbm.NewMemDB()}, nil, "", "", "", nil, nil) + app, _, _, cleanup, err := abciapp.NewDIDPLCApplication(t.Context(), logger, nil, dbm.NewMemDB(), memDBWrapper{dbm.NewMemDB()}, nil, "", "", "", nil, nil, config.DefaultPLCConfig()) require.NoError(t, err) t.Cleanup(cleanup) diff --git a/abciapp/execution.go b/abciapp/execution.go index c123c8c..2e90d88 100644 --- a/abciapp/execution.go +++ b/abciapp/execution.go @@ -340,6 +340,8 @@ func (d *DIDPLCApplication) Commit(context.Context, *abcitypes.RequestCommit) (* return nil, stacktrace.Propagate(err) } + committedHeight := d.ongoingRead.Height() + for _, r := range d.lastProcessedProposalExecTxResults { for _, cb := range r.commitSideEffects { cb() @@ -349,6 +351,22 @@ func (d *DIDPLCApplication) Commit(context.Context, *abcitypes.RequestCommit) (* d.ongoingWrite = nil d.ongoingRead = nil + // Notify snapshot manager of new committed height (non-blocking) + select { + case d.snapshotManagerLatestHeightChan <- committedHeight: + default: + // try updating the height buffered in the chan + + // tentative read because we could race at reading with the snapshot manager goroutine + select { + case <-d.snapshotManagerLatestHeightChan: + default: + } + + // this must succeed because no other goroutine writes to the chan, and we just read: + d.snapshotManagerLatestHeightChan <- committedHeight + } + return &abcitypes.ResponseCommit{ // TODO only discard actual blockchain history based on settings //RetainHeight: minHeightToKeep, diff --git a/abciapp/snapshots.go b/abciapp/snapshots.go index 51372ec..dff0a1b 100644 --- a/abciapp/snapshots.go +++ b/abciapp/snapshots.go @@ -15,6 +15,51 @@ import ( "tangled.org/gbl08ma.com/didplcbft/store" ) +// runSnapshotManager monitors block heights and creates/deletes snapshots as needed +func (d *DIDPLCApplication) runSnapshotManager(ctx context.Context, snapshotDirectory string) { + var lastHeight int64 + + for { + select { + case <-ctx.Done(): + return + case height := <-d.snapshotManagerLatestHeightChan: + lastHeight = height + } + + currentHeight := lastHeight + snapshotInterval := int64(d.plcConfig.SnapshotInterval) + + // Skip if we haven't reached the interval yet + if currentHeight-d.lastSnapshotHeight < snapshotInterval { + continue + } + + tempFilename := filepath.Join(snapshotDirectory, fmt.Sprintf("%020d.snapshot.tmp", currentHeight)) + + d.logger.Info("Creating snapshot", "height", currentHeight) + + err := d.createSnapshot(currentHeight, tempFilename) + if err != nil { + d.logger.Error("failed to create snapshot", "height", currentHeight, "error", stacktrace.Propagate(err)) + continue + } + + d.logger.Info("Created snapshot", "height", currentHeight) + d.lastSnapshotHeight = currentHeight + + // Clean up old snapshots if retention is set + if d.plcConfig.SnapshotRetentionCount > 0 { + numDeleted, err := store.Snapshot.PruneOldSnapshots(snapshotDirectory, int(d.plcConfig.SnapshotRetentionCount)) + if err != nil { + d.logger.Error("failed to prune old snapshots", "error", stacktrace.Propagate(err)) + continue + } + d.logger.Info("Pruned old snapshots", "numDeleted", numDeleted) + } + } +} + // snapshotNumRecentBlockHeaders is the number of recent block headers to include in snapshots // It should be enough to handle challenges that depend on recent block headers const snapshotNumRecentBlockHeaders = max(CommitToChallengeMaxAgeInBlocks, CompleteChallengeMaxAgeInBlocks) + 5 diff --git a/config/config.go b/config/config.go index 6c661ef..c7b0a97 100644 --- a/config/config.go +++ b/config/config.go @@ -32,6 +32,15 @@ type PLCConfig struct { // Server response timeout for API endpoints ResponseTimeout time.Duration `mapstructure:"response_timeout"` + + // SnapshotInterval defines the number of blocks between automatic snapshots. + // If set to 0, automatic snapshot creation is disabled. + SnapshotInterval uint64 `mapstructure:"snapshot_interval"` + + // SnapshotRetentionCount defines the maximum number of snapshots to retain. + // When creating a new snapshot, older snapshots are automatically deleted. + // If set to 0, all snapshots are retained. + SnapshotRetentionCount uint64 `mapstructure:"snapshot_retention_count"` } func DefaultPLCConfig() *PLCConfig { @@ -40,5 +49,7 @@ func DefaultPLCConfig() *PLCConfig { Pprof: true, // TODO set to false once we move past alpha phase MaxStreamingExportCursorAge: 7 * 24 * time.Hour, ResponseTimeout: 10 * time.Second, + SnapshotInterval: 0, // Disable snapshots by default + SnapshotRetentionCount: 1, } } diff --git a/main.go b/main.go index 55977e6..a5e0b69 100644 --- a/main.go +++ b/main.go @@ -155,7 +155,8 @@ func main() { cfg.StateSync.TempDir, didBloomFilterPath, mempoolSubmitter, - blockHeaderGetter) + blockHeaderGetter, + cfg.PLC) if err != nil { log.Fatalf("failed to create DIDPLC application: %v", err) } diff --git a/store/snapshot.go b/store/snapshot.go index e061b74..591fc1f 100644 --- a/store/snapshot.go +++ b/store/snapshot.go @@ -828,3 +828,70 @@ func (s *SnapshotStore) LoadSnapshotChunk(snapshotDirectory string, height uint6 return chunkData, nil } + +func (s *SnapshotStore) PruneOldSnapshots(snapshotDirectory string, retentionCount int) (int, error) { + // Get list of all snapshots sorted by height (newest first) + files, err := filepath.Glob(filepath.Join(snapshotDirectory, "*.snapshot")) + if err != nil { + return 0, stacktrace.Propagate(err, "failed to list snapshots") + } + + if len(files) <= retentionCount { + return 0, nil + } + + // Extract heights from filenames and sort + heights := make([]uint64, 0, len(files)) + for _, f := range files { + base := filepath.Base(f) + heightStr := strings.TrimSuffix(base, ".snapshot") + h, err := strconv.ParseUint(heightStr, 10, 64) + if err != nil { + continue + } + heights = append(heights, h) + } + + slices.SortFunc(heights, func(a, b uint64) int { + return int(int64(b) - int64(a)) // Sort descending (newest first) + }) + + // Delete snapshots beyond retention count + toDelete := heights[retentionCount:] + for _, h := range toDelete { + snapshotFile := filepath.Join(snapshotDirectory, fmt.Sprintf("%020d.snapshot", h)) + chunksumsFile := filepath.Join(snapshotDirectory, fmt.Sprintf("%020d.chunksums", h)) + + if err := os.Remove(snapshotFile); err != nil && !errors.Is(err, os.ErrNotExist) { + return 0, stacktrace.Propagate(err, "failed to delete old snapshot file: %s", snapshotFile) + } + + if err := os.Remove(chunksumsFile); err != nil && !errors.Is(err, os.ErrNotExist) { + return 0, stacktrace.Propagate(err, "failed to delete old chunksums file: %s", chunksumsFile) + } + } + + return len(toDelete), nil +} + +func (s *SnapshotStore) MostRecentSnapshotHeight(snapshotDirectory string) (uint64, error) { + files, err := filepath.Glob(filepath.Join(snapshotDirectory, "*.snapshot")) + if err != nil { + return 0, stacktrace.Propagate(err, "failed to list snapshots") + } + + var maxHeight uint64 + for _, f := range files { + base := filepath.Base(f) + heightStr := strings.TrimSuffix(base, ".snapshot") + h, err := strconv.ParseUint(heightStr, 10, 64) + if err != nil { + continue + } + if h > maxHeight { + maxHeight = h + } + } + + return maxHeight, nil +}