gs: Do not schedule too far ahead

Author: halimi

config/messages.json

@@ -5687,6 +5687,15 @@
       "file": "scheduler.go"
     }
   },
+  "error:pkg/gatewayserver/scheduling:schedule_too_far_ahead": {
+    "translations": {
+      "en": "schedule time is too far in the future"
+    },
+    "description": {
+      "package": "pkg/gatewayserver/scheduling",
+      "file": "sub_band.go"
+    }
+  },
   "error:pkg/gatewayserver/scheduling:sub_band_not_found": {
     "translations": {
       "en": "sub-band not found for frequency `{frequency}` Hz"

pkg/gatewayserver/scheduling/scheduler.go

@@ -459,7 +459,7 @@ func (s *Scheduler) ScheduleAnytime(ctx context.Context, opts Options) (res Emis
 		}
 		return em.t
 	}
-	em, err = sb.ScheduleAnytime(em.d, next, opts.Priority)
+	em, err = sb.ScheduleAnytime(em.d, next, opts.Priority, now)
 	if err != nil {
 		return Emission{}, 0, err
 	}

pkg/gatewayserver/scheduling/scheduler_internal_test.go

@@ -15,8 +15,9 @@
 package scheduling
 
 var (
-	ErrConflict  = errConflict
-	ErrDwellTime = errDwellTime
-	ErrTooLate   = errTooLate
-	ErrDutyCycle = errDutyCycle
+	ErrConflict            = errConflict
+	ErrDwellTime           = errDwellTime
+	ErrTooLate             = errTooLate
+	ErrDutyCycle           = errDutyCycle
+	ErrScheduleTooFarAhead = errScheduleTooFarAhead
 )

pkg/gatewayserver/scheduling/sub_band.go

@@ -27,6 +27,11 @@ import (
 // A lower value results in balancing capacity in time, while a higher value allows for bursts.
 var DutyCycleWindow = 1 * time.Hour
 
+// MaxScheduleAhead is the maximum duration into the future that ScheduleAnytime may place an emission.
+// Emissions beyond this are rejected to avoid scheduling downlinks that would never actually be transmitted
+// in practice.
+var MaxScheduleAhead = 2 * DutyCycleWindow
+
 // DutyCycleStyle represents the of duty cycle algorithm to be used by a sub band.
 type DutyCycleStyle int
 
@@ -144,6 +149,10 @@ var (
 		"blocked",
 		"sub band is blocked for `{duration}`",
 	)
+	errScheduleTooFarAhead = errors.DefineResourceExhausted(
+		"schedule_too_far_ahead",
+		"schedule time is too far in the future",
+	)
 )
 
 // checkDutyCycle verifies if the emission complies with the duty cycle limitations, based on the style.
@@ -211,7 +220,12 @@ func (sb *SubBand) Schedule(em Emission, p ttnpb.TxSchedulePriority) error {
 // ScheduleAnytime schedules the given duration at a time when there is availability by accounting for duty-cycle.
 // The given next callback should return the next option that does not conflict with other scheduled downlinks.
 // If there is no duty-cycle limitation, this method returns the first option.
-func (sb *SubBand) ScheduleAnytime(d time.Duration, next func() ConcentratorTime, p ttnpb.TxSchedulePriority) (Emission, error) {
+func (sb *SubBand) ScheduleAnytime(
+	d time.Duration,
+	next func() ConcentratorTime,
+	p ttnpb.TxSchedulePriority,
+	now ConcentratorTime,
+) (Emission, error) {
 	sb.mu.Lock()
 	defer sb.mu.Unlock()
 	em := NewEmission(next(), d)
@@ -241,7 +255,13 @@ func (sb *SubBand) ScheduleAnytime(d time.Duration, next func() ConcentratorTime
 			other := sb.emissions[i]
 			used += float32(other.d) / float32(DutyCycleWindow)
 			if used > usable {
-				em.t = other.Ends() + ConcentratorTime(DutyCycleWindow) - ConcentratorTime(em.d)
+				newT := other.Ends() + ConcentratorTime(DutyCycleWindow) - ConcentratorTime(em.d)
+				// If the new time is too far in the future, return an error instead of scheduling an emission that would
+				// never be transmitted in practice.
+				if newT > now+ConcentratorTime(MaxScheduleAhead) {
+					return Emission{}, errScheduleTooFarAhead.New()
+				}
+				em.t = newT
 				break
 			}
 		}

pkg/gatewayserver/scheduling/sub_band_test.go

@@ -172,6 +172,8 @@ func TestSubBandScheduleRestricted(t *testing.T) {
 }
 
 func TestScheduleAnytimeRestricted(t *testing.T) {
+	t.Parallel()
+
 	a := assertions.New(t)
 	params := scheduling.SubBandParameters{
 		MinFrequency: 0,
@@ -204,7 +206,8 @@ func TestScheduleAnytimeRestricted(t *testing.T) {
 			from += scheduling.ConcentratorTime(time.Second)
 			return res
 		}
-		em, err := sb.ScheduleAnytime(time.Second, next, ttnpb.TxSchedulePriority_NORMAL)
+		now := scheduling.ConcentratorTime(time.Now().UnixNano())
+		em, err := sb.ScheduleAnytime(time.Second, next, ttnpb.TxSchedulePriority_NORMAL, now)
 		a.So(err, should.BeNil)
 		a.So(em.Starts(), should.Equal, 16*time.Second)
 		// [     1       2 4 3        ]
@@ -217,7 +220,8 @@ func TestScheduleAnytimeRestricted(t *testing.T) {
 		next := func() scheduling.ConcentratorTime {
 			return scheduling.ConcentratorTime(19 * time.Second)
 		}
-		em, err := sb.ScheduleAnytime(time.Second, next, ttnpb.TxSchedulePriority_NORMAL)
+		now := scheduling.ConcentratorTime(time.Now().UnixNano())
+		em, err := sb.ScheduleAnytime(time.Second, next, ttnpb.TxSchedulePriority_NORMAL, now)
 		a.So(err, should.BeNil)
 		a.So(em.Starts(), should.Equal, 26*time.Second)
 		// [     1       2 4 3       5]
@@ -229,11 +233,60 @@ func TestScheduleAnytimeRestricted(t *testing.T) {
 		next := func() scheduling.ConcentratorTime {
 			return scheduling.ConcentratorTime(19 * time.Second)
 		}
-		_, err := sb.ScheduleAnytime(5*time.Second, next, ttnpb.TxSchedulePriority_NORMAL)
+		now := scheduling.ConcentratorTime(time.Now().UnixNano())
+		_, err := sb.ScheduleAnytime(5*time.Second, next, ttnpb.TxSchedulePriority_NORMAL, now)
 		a.So(err, should.HaveSameErrorDefinitionAs, scheduling.ErrDutyCycle)
 	}
 }
 
+func TestScheduleAnytimeTooFarAhead(t *testing.T) {
+	t.Parallel()
+	// In the test environment DutyCycleWindow = 10s and MaxScheduleAhead = 20s.
+	a := assertions.New(t)
+	params := scheduling.SubBandParameters{
+		MinFrequency: 0,
+		MaxFrequency: math.MaxUint64,
+		DutyCycle:    0.5,
+	}
+	clock := &mockClock{}
+	ceilings := map[ttnpb.TxSchedulePriority]float32{
+		ttnpb.TxSchedulePriority_NORMAL:  0.5, // usable = 0.25
+		ttnpb.TxSchedulePriority_HIGHEST: 1.0, // usable = 0.50
+	}
+	sb := scheduling.NewSubBand(params, clock, ceilings, scheduling.DefaultDutyCycleStyle)
+
+	// Schedule a 3-second emission at t=9s using HIGHEST priority.
+	// It occupies 30% of the duty-cycle window (10s), within HIGHEST (50%) but above
+	// NORMAL (25%). Its window [9s, 12s] falls within the checkDutyCycle range
+	// [0, 10s] for an emission starting at t=0, causing that check to fail.
+	err := sb.Schedule(
+		scheduling.NewEmission(scheduling.ConcentratorTime(9*time.Second), 3*time.Second),
+		ttnpb.TxSchedulePriority_HIGHEST,
+	)
+	a.So(err, should.BeNil)
+
+	// next() always returns the same ConcentratorTime, so ScheduleAnytime falls back
+	// to the backwards scan. newT = 9s + 3s + 10s - 1s = 21s, which exceeds
+	// now (0) + MaxScheduleAhead (20s).
+	{
+		next := func() scheduling.ConcentratorTime { return 0 }
+		now := scheduling.ConcentratorTime(0)
+		_, err := sb.ScheduleAnytime(time.Second, next, ttnpb.TxSchedulePriority_NORMAL, now)
+		a.So(err, should.HaveSameErrorDefinitionAs, scheduling.ErrScheduleTooFarAhead)
+	}
+
+	// With now shifted forward by 1 second, newT (21s) equals now + MaxScheduleAhead
+	// (1s + 20s = 21s) exactly, so the check (strictly greater than) does not fire
+	// and the emission is accepted.
+	{
+		next := func() scheduling.ConcentratorTime { return 0 }
+		now := scheduling.ConcentratorTime(time.Second)
+		em, err := sb.ScheduleAnytime(time.Second, next, ttnpb.TxSchedulePriority_NORMAL, now)
+		a.So(err, should.BeNil)
+		a.So(em.Starts(), should.Equal, 21*time.Second)
+	}
+}
+
 func TestBlockingScheduling(t *testing.T) {
 	t.Parallel()
 

pkg/gatewayserver/scheduling/util_test.go

@@ -80,4 +80,5 @@ func float32Ptr(v float32) *float32              { return &v }
 
 func init() {
 	scheduling.DutyCycleWindow = 10 * time.Second
+	scheduling.MaxScheduleAhead = 2 * scheduling.DutyCycleWindow
 }

pkg/webui/locales/ja.json

@@ -2520,6 +2520,7 @@
   "error:pkg/gatewayserver/scheduling:no_absolute_gateway_time": "アブソリュートゲートウェイタイムがありません",
   "error:pkg/gatewayserver/scheduling:no_clock_sync": "クロックが未同期",
   "error:pkg/gatewayserver/scheduling:no_server_time": "サーバの時間がありません",
+  "error:pkg/gatewayserver/scheduling:schedule_too_far_ahead": "",
   "error:pkg/gatewayserver/scheduling:sub_band_not_found": "周波数 `{frequency}` Hz に対するサブ帯域が見つかりません",
   "error:pkg/gatewayserver/scheduling:too_late": "予定された時間（`{delta}`）で送信するには遅すぎます",
   "error:pkg/gatewayserver/upstream/ns:network_server_not_found": "ネットワークサーバが見つかりません",