ns: Retry confirmed class C downlink on timeout

Author: halimi

cmd/ttn-lw-stack/commands/start.go

@@ -53,7 +53,12 @@ import (
 	"go.thethings.network/lorawan-stack/v3/pkg/web"
 )
 
-const defaultLockTTL = 10 * time.Second
+const (
+	defaultLockTTL = 10 * time.Second
+
+	downlinkKey        = "downlink"
+	pendingDownlinkKey = "pending-downlink"
+)
 
 // NewComponentDeviceRegistryRedis instantiates a new redis client with the Component Device Registry namespace.
 func NewComponentDeviceRegistryRedis(conf *Config, name string) *redis.Client {
@@ -78,6 +83,12 @@ func NewNetworkServerDownlinkTaskRedis(conf *Config) *redis.Client {
 	return redis.New(conf.Redis.WithNamespace("ns", "tasks"))
 }
 
+// NewNetworkServerPendingDownlinkTaskRedis instantiates a new redis client
+// with the Network Server Pending Downlink Task namespace.
+func NewNetworkServerPendingDownlinkTaskRedis(conf *Config) *redis.Client {
+	return redis.New(conf.Redis.WithNamespace("ns", "pending-tasks"))
+}
+
 // NewNetworkServerMACSettingsProfileRegistryRedis instantiates a new redis client
 // with the Network Server MAC Settings Profile Registry namespace.
 func NewNetworkServerMACSettingsProfileRegistryRedis(conf *Config) *redis.Client {
@@ -372,12 +383,25 @@ var startCommand = &cobra.Command{
 				100000,
 				redisConsumerGroup,
 				redis.DefaultStreamBlockLimit,
+				downlinkKey,
 			)
 			if err := downlinkTasks.Init(ctx); err != nil {
 				return shared.ErrInitializeNetworkServer.WithCause(err)
 			}
 			defer downlinkTasks.Close(ctx)
 			config.NS.DownlinkTaskQueue.Queue = downlinkTasks
+			pendingDownlinkTasks := nsredis.NewDownlinkTaskQueue(
+				NewNetworkServerPendingDownlinkTaskRedis(config),
+				100000,
+				redisConsumerGroup,
+				redis.DefaultStreamBlockLimit,
+				pendingDownlinkKey,
+			)
+			if err := pendingDownlinkTasks.Init(ctx); err != nil {
+				return shared.ErrInitializeNetworkServer.WithCause(err)
+			}
+			defer pendingDownlinkTasks.Close(ctx)
+			config.NS.PendingDownlinkTaskQueue.Queue = pendingDownlinkTasks
 			config.NS.ScheduledDownlinkMatcher = &nsredis.ScheduledDownlinkMatcher{
 				Redis: redis.New(config.Cache.Redis.WithNamespace("ns", "scheduled-downlinks")),
 			}

pkg/networkserver/config.go

@@ -34,7 +34,7 @@ type ApplicationUplinkQueueConfig struct {
 	FastNumConsumers uint64 `name:"fast-num-consumers"`
 }
 
-// ApplicationUplinkQueueConfig defines downlink task queue configuration.
+// DownlinkTaskQueueConfig defines downlink task queue configuration.
 type DownlinkTaskQueueConfig struct {
 	Queue        DownlinkTaskQueue `name:"-"`
 	NumConsumers uint64            `name:"num-consumers"`
@@ -141,6 +141,7 @@ type Config struct {
 	ApplicationUplinkQueue     ApplicationUplinkQueueConfig `name:"application-uplink-queue"`
 	Devices                    DeviceRegistry               `name:"-"`
 	DownlinkTaskQueue          DownlinkTaskQueueConfig      `name:"downlink-task-queue"`
+	PendingDownlinkTaskQueue   DownlinkTaskQueueConfig      `name:"pending-downlink-task-queue"`
 	UplinkDeduplicator         UplinkDeduplicator           `name:"-"`
 	ScheduledDownlinkMatcher   ScheduledDownlinkMatcher     `name:"-"`
 	NetID                      types.NetID                  `name:"net-id" description:"NetID of this Network Server"`                                                                                   // nolint: lll
@@ -170,6 +171,9 @@ var DefaultConfig = Config{
 	DownlinkTaskQueue: DownlinkTaskQueueConfig{
 		NumConsumers: 1,
 	},
+	PendingDownlinkTaskQueue: DownlinkTaskQueueConfig{
+		NumConsumers: 1,
+	},
 	DeduplicationWindow: 200 * time.Millisecond,
 	CooldownWindow:      time.Second,
 	DownlinkPriorities: DownlinkPriorityConfig{

pkg/networkserver/downlink.go

@@ -1473,7 +1473,15 @@ type downlinkAttemptResult struct {
 	DownlinkTaskUpdateStrategy downlinkTaskUpdateStrategy
 }
 
-func (ns *NetworkServer) attemptClassADataDownlink(ctx context.Context, dev *ttnpb.EndDevice, phy *band.Band, fp *frequencyplans.FrequencyPlan, slot *classADownlinkSlot, maxUpLength uint16) downlinkAttemptResult {
+func (ns *NetworkServer) attemptClassADataDownlink( // nolint:gocyclo
+	ctx context.Context,
+	dev *ttnpb.EndDevice,
+	phy *band.Band,
+	fp *frequencyplans.FrequencyPlan,
+	slot *classADownlinkSlot,
+	maxUpLength uint16,
+	profile *ttnpb.MACSettings,
+) downlinkAttemptResult {
 	ctx = events.ContextWithCorrelationID(ctx, slot.Uplink.CorrelationIds...)
 	if !dev.MacState.RxWindowsAvailable {
 		log.FromContext(ctx).Error("RX windows not available, skip class A downlink slot")
@@ -1661,6 +1669,16 @@ func (ns *NetworkServer) attemptClassADataDownlink(ctx context.Context, dev *ttn
 		dev.Session.QueuedApplicationDownlinks = dev.Session.QueuedApplicationDownlinks[:0:0]
 	}
 	recordDataDownlink(dev, genState, genDown.NeedsMACAnswer, down, ns.defaultMACSettings)
+	if genState.ApplicationDownlink != nil &&
+		genState.ApplicationDownlink.Confirmed &&
+		dev.MacState.DeviceClass == ttnpb.Class_CLASS_C {
+		timeout := mac.DeviceClassCTimeout(dev, ns.defaultMACSettings, profile)
+		taskAt := time.Now().UTC().Add(timeout)
+		log.FromContext(ctx).WithField("start_at", taskAt).Debug("Add pending downlink task")
+		if err := ns.pendingDownlinkTasks.Add(ctx, dev.Ids, taskAt, true); err != nil {
+			log.FromContext(ctx).WithError(err).Warn("Failed to add pending downlink task")
+		}
+	}
 	dev.MacState.PendingRelayDownlink = nil
 	return downlinkAttemptResult{
 		SetPaths: ttnpb.AddFields(sets,
@@ -1679,7 +1697,15 @@ func (ns *NetworkServer) attemptClassADataDownlink(ctx context.Context, dev *ttn
 	}
 }
 
-func (ns *NetworkServer) attemptNetworkInitiatedDataDownlink(ctx context.Context, dev *ttnpb.EndDevice, phy *band.Band, fp *frequencyplans.FrequencyPlan, slot *networkInitiatedDownlinkSlot, maxUpLength uint16) downlinkAttemptResult {
+func (ns *NetworkServer) attemptNetworkInitiatedDataDownlink( // nolint:gocyclo
+	ctx context.Context,
+	dev *ttnpb.EndDevice,
+	phy *band.Band,
+	fp *frequencyplans.FrequencyPlan,
+	slot *networkInitiatedDownlinkSlot,
+	maxUpLength uint16,
+	profile *ttnpb.MACSettings,
+) downlinkAttemptResult {
 	var drIdx ttnpb.DataRateIndex
 	var freq uint64
 	switch slot.Class {
@@ -1890,6 +1916,16 @@ func (ns *NetworkServer) attemptNetworkInitiatedDataDownlink(ctx context.Context
 	}
 
 	recordDataDownlink(dev, genState, genDown.NeedsMACAnswer, down, ns.defaultMACSettings)
+	if genState.ApplicationDownlink != nil &&
+		genState.ApplicationDownlink.Confirmed &&
+		dev.MacState.DeviceClass == ttnpb.Class_CLASS_C {
+		timeout := mac.DeviceClassCTimeout(dev, ns.defaultMACSettings, profile)
+		taskAt := time.Now().UTC().Add(timeout)
+		log.FromContext(ctx).WithField("start_at", taskAt).Debug("Add pending downlink task")
+		if err := ns.pendingDownlinkTasks.Add(ctx, dev.Ids, taskAt, true); err != nil {
+			log.FromContext(ctx).WithError(err).Warn("Failed to add pending downlink task")
+		}
+	}
 	if genState.ApplicationDownlink != nil || genState.EvictDownlinkQueueIfScheduled {
 		sets = ttnpb.AddFields(sets, "session.queued_application_downlinks")
 	}
@@ -2198,7 +2234,7 @@ func (ns *NetworkServer) processDownlinkTask(ctx context.Context, consumerID str
 					}
 					switch slot := v.(type) {
 					case *classADownlinkSlot:
-						a := ns.attemptClassADataDownlink(ctx, dev, phy, fp, slot, maxUpLength)
+						a := ns.attemptClassADataDownlink(ctx, dev, phy, fp, slot, maxUpLength, profile.GetMacSettings())
 						queuedEvents = append(queuedEvents, a.QueuedEvents...)
 						queuedApplicationUplinks = append(queuedApplicationUplinks, a.QueuedApplicationUplinks...)
 						taskUpdateStrategy = a.DownlinkTaskUpdateStrategy
@@ -2229,7 +2265,7 @@ func (ns *NetworkServer) processDownlinkTask(ctx context.Context, consumerID str
 							earliestAt = time.Now().Add(absoluteTimeSchedulingDelay / 2)
 							continue
 						}
-						a := ns.attemptNetworkInitiatedDataDownlink(ctx, dev, phy, fp, slot, maxUpLength)
+						a := ns.attemptNetworkInitiatedDataDownlink(ctx, dev, phy, fp, slot, maxUpLength, profile.GetMacSettings())
 						queuedEvents = append(queuedEvents, a.QueuedEvents...)
 						queuedApplicationUplinks = append(queuedApplicationUplinks, a.QueuedApplicationUplinks...)
 						taskUpdateStrategy = a.DownlinkTaskUpdateStrategy
@@ -2273,3 +2309,90 @@ func (ns *NetworkServer) processDownlinkTask(ctx context.Context, consumerID str
 	}
 	return err
 }
+
+const maxConfirmedDownlinkRetries = 8
+
+func (ns *NetworkServer) createProcessPendingDownlinkTask(consumerID string) func(context.Context) error {
+	return func(ctx context.Context) error {
+		return ns.processPendingDownlinkTask(ctx, consumerID)
+	}
+}
+
+// processPendingDownlinkTask processes the most recent pending downlink task ready for execution,
+// if such is available or wait until it is before processing it.
+// NOTE: ctx.Done() is not guaranteed to be respected by processPendingDownlinkTask.
+// The processPendingDownlinkTask receives the consumerID that will be used for popping
+// from the pending downlink task queue.
+func (ns *NetworkServer) processPendingDownlinkTask(ctx context.Context, consumerID string) error {
+	var setErr bool
+	err := ns.pendingDownlinkTasks.Pop(
+		ctx,
+		consumerID,
+		func(ctx context.Context, devID *ttnpb.EndDeviceIdentifiers, t time.Time,
+		) (time.Time, error) {
+			ctx = log.NewContextWithFields(ctx, log.Fields(
+				"device_uid", unique.ID(ctx, devID),
+				"started_at", time.Now().UTC(),
+			))
+			logger := log.FromContext(ctx)
+			logger.WithField("start_at", t).Debug("Process pending downlink task")
+
+			dev, ctx, err := ns.devices.SetByID(ctx, devID.ApplicationIds, devID.DeviceId,
+				[]string{
+					"mac_state",
+					"session",
+				},
+				func(_ context.Context, dev *ttnpb.EndDevice) (*ttnpb.EndDevice, []string, error) {
+					if dev == nil {
+						logger.Warn("Device not found")
+						return nil, nil, nil
+					}
+
+					pendingAppDown := dev.MacState.GetPendingApplicationDownlink()
+					if pendingAppDown != nil {
+						pendingAppDown.FCnt = dev.Session.LastNFCntDown + 1
+						pendingAppDown.ConfirmedRetry.Attempt++
+
+						if pendingAppDown.ConfirmedRetry.Attempt > maxConfirmedDownlinkRetries {
+							dev.MacState.PendingApplicationDownlink = nil
+							logger.Warn("Max confirmed downlink retries reached, drop pending application downlink")
+							return dev, []string{
+								"mac_state.pending_application_downlink",
+							}, nil
+						}
+
+						// Enqueue the pending application downlink at the front of the queue.
+						// This preserves the order of other queued application downlinks.
+						// The pending application downlink will be processed first in the next downlink task.
+						// This is important for confirmed downlinks, as we need to ensure that the ACK is received
+						// before processing other downlinks.
+						dev.Session.QueuedApplicationDownlinks = append(
+							[]*ttnpb.ApplicationDownlink{pendingAppDown},
+							dev.Session.QueuedApplicationDownlinks...,
+						)
+						dev.MacState.PendingApplicationDownlink = nil
+					}
+					return dev, []string{
+						"mac_state",
+						"session",
+					}, nil
+				},
+			)
+			if err != nil {
+				setErr = true
+				logger.WithError(err).Error("Failed to update device in registry")
+				return time.Time{}, err
+			}
+
+			if err := ns.updateDataDownlinkTask(ctx, dev, time.Time{}); err != nil {
+				log.FromContext(ctx).WithError(err).Error(
+					"Failed to update downlink task queue after processing pending downlink")
+			}
+
+			return time.Time{}, nil
+		})
+	if err != nil && !setErr {
+		log.FromContext(ctx).WithError(err).Error("Failed to pop entry from pending downlink task queue")
+	}
+	return err
+}

pkg/networkserver/internal/test/shared/redis.go

@@ -70,7 +70,7 @@ func NewRedisDeviceRegistry(ctx context.Context) (DeviceRegistry, func()) {
 func NewRedisDownlinkTaskQueue(ctx context.Context) (DownlinkTaskQueue, func()) {
 	tb := test.MustTBFromContext(ctx)
 	cl, flush := test.NewRedis(ctx, append(redisNamespace[:], "downlink-tasks")...)
-	q := redis.NewDownlinkTaskQueue(cl, 10000, redisConsumerGroup, testStreamBlockLimit())
+	q := redis.NewDownlinkTaskQueue(cl, 10000, redisConsumerGroup, testStreamBlockLimit(), "downlink")
 	if err := q.Init(ctx); err != nil {
 		tb.Fatalf("Failed to initialize Redis downlink task queue: %s", test.FormatError(err))
 	}

pkg/networkserver/networkserver.go

@@ -172,8 +172,9 @@ type NetworkServer struct {
 
 	applicationUplinks ApplicationUplinkQueue
 
-	downlinkTasks      DownlinkTaskQueue
-	downlinkPriorities DownlinkPriorities
+	downlinkTasks        DownlinkTaskQueue
+	downlinkPriorities   DownlinkPriorities
+	pendingDownlinkTasks DownlinkTaskQueue
 
 	deduplicationWindow windowDurationFunc
 	collectionWindow    windowDurationFunc
@@ -211,6 +212,7 @@ const (
 	downlinkProcessTaskName           = "process_downlink"
 	applicationUplinkDispatchTaskName = "dispatch_application_uplink"
 	downlinkDispatchTaskName          = "dispatch_downlink"
+	pendingDownlinkProcessTaskName    = "process_pending_downlink"
 
 	maxInt = int(^uint(0) >> 1)
 )
@@ -230,10 +232,14 @@ func New(c *component.Component, conf *Config, opts ...Option) (*NetworkServer,
 		panic(errInvalidConfiguration.WithCause(errors.New("Devices is not specified")))
 	case conf.DownlinkTaskQueue.NumConsumers == 0:
 		return nil, errInvalidConfiguration.WithCause(errors.New("DownlinkTaskQueue.NumConsumers must be greater than 0"))
+	case conf.PendingDownlinkTaskQueue.NumConsumers == 0:
+		return nil, errInvalidConfiguration.WithCause(errors.New("PendingDownlinkTaskQueue.NumConsumers must be greater than 0")) // nolint:lll
 	case conf.ApplicationUplinkQueue.NumConsumers == 0:
 		return nil, errInvalidConfiguration.WithCause(errors.New("ApplicationUplinkQueue.NumConsumers must be greater than 0"))
 	case conf.DownlinkTaskQueue.Queue == nil:
 		panic(errInvalidConfiguration.WithCause(errors.New("DownlinkTaskQueue is not specified")))
+	case conf.PendingDownlinkTaskQueue.Queue == nil:
+		panic(errInvalidConfiguration.WithCause(errors.New("PendingDownlinkTaskQueue is not specified")))
 	case conf.UplinkDeduplicator == nil:
 		panic(errInvalidConfiguration.WithCause(errors.New("UplinkDeduplicator is not specified")))
 	case conf.ScheduledDownlinkMatcher == nil:
@@ -297,6 +303,7 @@ func New(c *component.Component, conf *Config, opts ...Option) (*NetworkServer,
 		macSettingsProfile:       &NsMACSettingsProfileRegistry{registry: conf.MACSettingsProfileRegistry},
 		macSettingsProfiles:      conf.MACSettingsProfileRegistry,
 		downlinkTasks:            conf.DownlinkTaskQueue.Queue,
+		pendingDownlinkTasks:     conf.PendingDownlinkTaskQueue.Queue,
 		downlinkPriorities:       downlinkPriorities,
 		defaultMACSettings:       defaultMACSettings,
 		interopClient:            interopCl,
@@ -358,7 +365,8 @@ func New(c *component.Component, conf *Config, opts ...Option) (*NetworkServer,
 	for id, dispatcher := range map[string]interface {
 		Dispatch(context.Context, string) error
 	}{
-		downlinkDispatchTaskName: ns.downlinkTasks,
+		downlinkDispatchTaskName:       ns.downlinkTasks,
+		pendingDownlinkProcessTaskName: ns.pendingDownlinkTasks,
 	} {
 		dispatcher := dispatcher
 		ns.RegisterTask(&task.Config{
@@ -391,6 +399,16 @@ func New(c *component.Component, conf *Config, opts ...Option) (*NetworkServer,
 			Backoff: processTaskBackoff,
 		})
 	}
+	for i := uint64(0); i < conf.PendingDownlinkTaskQueue.NumConsumers; i++ {
+		consumerID := fmt.Sprintf("%s:%d", consumerIDPrefix, i)
+		ns.RegisterTask(&task.Config{
+			Context: ctx,
+			ID:      fmt.Sprintf("%s_%d", pendingDownlinkProcessTaskName, i),
+			Func:    ns.createProcessPendingDownlinkTask(consumerID),
+			Restart: task.RestartAlways,
+			Backoff: processTaskBackoff,
+		})
+	}
 	c.RegisterGRPC(ns)
 	return ns, nil
 }

pkg/networkserver/redis/downlink_task_queue.go

@@ -29,20 +29,16 @@ type DownlinkTaskQueue struct {
 	queue *ttnredis.TaskQueue
 }
 
-const (
-	downlinkKey = "downlink"
-)
-
 // NewDownlinkTaskQueue returns new downlink task queue.
 func NewDownlinkTaskQueue(
-	cl *ttnredis.Client, maxLen int64, group string, streamBlockLimit time.Duration,
+	cl *ttnredis.Client, maxLen int64, group string, streamBlockLimit time.Duration, key string,
 ) *DownlinkTaskQueue {
 	return &DownlinkTaskQueue{
 		queue: &ttnredis.TaskQueue{
 			Redis:            cl,
 			MaxLen:           maxLen,
 			Group:            group,
-			Key:              cl.Key(downlinkKey),
+			Key:              cl.Key(key),
 			StreamBlockLimit: streamBlockLimit,
 		},
 	}