fix: in spanner and postgres, use min or max checkpointintervals

Author: miparnisari

internal/datastore/crdb/watch.go

@@ -78,7 +78,6 @@ func (cds *crdbDatastore) DefaultsWatchOptions() datastore.WatchOptions {
 }
 
 func (cds *crdbDatastore) Watch(ctx context.Context, afterRevision datastore.Revision, options datastore.WatchOptions) (<-chan datastore.RevisionChanges, <-chan error) {
-	fmt.Printf("%+v\n", options)
 	updates := make(chan datastore.RevisionChanges, options.WatchBufferLength)
 	errs := make(chan error, 1)
 

internal/datastore/postgres/watch.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/authzed/spicedb/internal/datastore/common"
 	"github.com/authzed/spicedb/internal/datastore/postgres/schema"
+	log "github.com/authzed/spicedb/internal/logging"
 	"github.com/authzed/spicedb/internal/sharederrors"
 	"github.com/authzed/spicedb/pkg/datastore"
 	core "github.com/authzed/spicedb/pkg/proto/core/v1"
@@ -26,7 +27,6 @@ const (
 
 func (pgd *pgDatastore) DefaultsWatchOptions() datastore.WatchOptions {
 	return datastore.WatchOptions{
-		CheckpointInterval:      minimumWatchSleep,
 		WatchBufferLength:       defaultWatchBufferLength,
 		WatchBufferWriteTimeout: defaultWatchBufferWriteTimeout,
 		// Postgres does not use WatchConnectTimeout
@@ -94,6 +94,11 @@ func (pgd *pgDatastore) Watch(
 
 	afterRevision := afterRevisionRaw.(postgresRevision)
 
+	if options.CheckpointInterval < minimumWatchSleep {
+		log.Warn().Msgf("--watch-api-heartbeat set too small, using %d", minimumWatchSleep)
+		options.CheckpointInterval = minimumWatchSleep
+	}
+
 	sendChange := func(change datastore.RevisionChanges) bool {
 		select {
 		case updates <- change:

internal/datastore/spanner/options.go

@@ -73,6 +73,8 @@ const (
 	defaultRevisionQuantization        = 5 * time.Second
 	defaultFollowerReadDelay           = 0 * time.Second
 	defaultMaxRevisionStalenessPercent = 0.1
+	minimumCheckpointInterval          = 100 * time.Millisecond
+	maximumCheckpointInterval          = 300000 * time.Millisecond
 	defaultWatchBufferLength           = 128
 	defaultWatchBufferWriteTimeout     = 1 * time.Second
 	defaultDisableStats                = false

internal/datastore/spanner/watch.go

@@ -18,6 +18,7 @@ import (
 
 	"github.com/authzed/spicedb/internal/datastore/common"
 	"github.com/authzed/spicedb/internal/datastore/revisions"
+	log "github.com/authzed/spicedb/internal/logging"
 	"github.com/authzed/spicedb/pkg/datastore"
 	core "github.com/authzed/spicedb/pkg/proto/core/v1"
 	"github.com/authzed/spicedb/pkg/spiceerrors"
@@ -55,7 +56,6 @@ func parseDatabaseName(db string) (project, instance, database string, err error
 
 func (sd *spannerDatastore) DefaultsWatchOptions() datastore.WatchOptions {
 	return datastore.WatchOptions{
-		CheckpointInterval:      100 * time.Millisecond,
 		WatchBufferLength:       defaultWatchBufferLength,
 		WatchBufferWriteTimeout: defaultWatchBufferWriteTimeout,
 		// Spanner does not use WatchConnectTimeout
@@ -88,6 +88,15 @@ func (sd *spannerDatastore) watch(
 	defer close(updates)
 	defer close(errs)
 
+	if opts.CheckpointInterval < minimumCheckpointInterval {
+		log.Warn().Msgf("--watch-api-heartbeat set too small, using %d", minimumCheckpointInterval)
+		opts.CheckpointInterval = minimumCheckpointInterval
+	}
+	if opts.CheckpointInterval > maximumCheckpointInterval {
+		log.Warn().Msgf("--watch-api-heartbeat set too high, using %d", maximumCheckpointInterval)
+		opts.CheckpointInterval = maximumCheckpointInterval
+	}
+
 	sendError := func(err error) {
 		if errors.Is(ctx.Err(), context.Canceled) || common.IsCancellationError(err) {
 			errs <- datastore.NewWatchCanceledErr()