Function FuzzRestoreWithMissingCompactedFile

restore_fuzz_test.go:17–135

(f *testing.F)

)

func FuzzRestoreWithMissingCompactedFile(f *testing.F) {
	f.Add(int64(1))
	f.Add(int64(2))
	f.Add(int64(3))

	f.Fuzz(func(t *testing.T, seed int64) {
		if testing.Short() {
			t.Skip("skipping fuzz test in short mode")
		}

		rng := rand.New(rand.NewSource(seed))
		ctx := t.Context()

		db := testingutil.NewDB(t, filepath.Join(t.TempDir(), "db"))
		db.MonitorInterval = 20 * time.Millisecond
		db.Replica = litestream.NewReplica(db)
		db.Replica.SyncInterval = 20 * time.Millisecond
		client := file.NewReplicaClient(t.TempDir())
		db.Replica.Client = client

		if err := db.Open(); err != nil {
			t.Fatal(err)
		}
		sqldb := testingutil.MustOpenSQLDB(t, db.Path())
		defer testingutil.MustCloseDBs(t, db, sqldb)

		store := litestream.NewStore([]*litestream.DB{db}, litestream.CompactionLevels{
			{Level: 0},
			{Level: 1, Interval: 50 * time.Millisecond},
			{Level: 2, Interval: 150 * time.Millisecond},
		})
		store.SnapshotInterval = 200 * time.Millisecond
		if err := store.Open(ctx); err != nil {
			t.Fatal(err)
		}
		defer store.Close(ctx)

		if _, err := sqldb.ExecContext(ctx, `CREATE TABLE t (id INTEGER PRIMARY KEY, val TEXT);`); err != nil {
			t.Fatal(err)
		}

		done := make(chan struct{})
		var wg sync.WaitGroup
		wg.Add(1)
		go func() {
			defer wg.Done()
			ticker := time.NewTicker(5 * time.Millisecond)
			defer ticker.Stop()

			for {
				select {
				case <-ctx.Done():
					return
				case <-done:
					return
				case <-ticker.C:
					if _, err := sqldb.ExecContext(ctx, `INSERT INTO t (val) VALUES (?);`, time.Now().String()); err != nil {
						return