| 952 | } |
| 953 | |
| 954 | pub async fn get_with_class_b_c_queue_items(limit: usize) -> Result<Vec<Device>> { |
| 955 | let mut c = get_async_db_conn().await?; |
| 956 | c.transaction::<Vec<Device>, Error, _>(async |c| { |
| 957 | let conf = config::get(); |
| 958 | |
| 959 | // This query will: |
| 960 | // * Select the devices for which a Class-B or Class-C downlink can be scheduled. |
| 961 | // * Lock the device records for update with skip locked such that other |
| 962 | // ChirpStack instances are able to do the same for the remaining devices. |
| 963 | // * Update the scheduler_run_after for these devices to now() + 2 * scheduler |
| 964 | // interval to avoid concurrency issues (other ChirpStack instance scheduling |
| 965 | // the same queue items). |
| 966 | // |
| 967 | // This way, we do not have to keep the device records locked until the scheduler |
| 968 | // finishes its batch as the same set of devices will not be returned until after |
| 969 | // the updated scheduler_run_after. Only if the scheduler takes more time than 2x the |
| 970 | // interval (the scheduler is still working on processing the batch after 2 x interval) |
| 971 | // this might cause issues. |
| 972 | // The alternative would be to keep the transaction open for a long time + keep |
| 973 | // the device records locked during this time which could case issues as well. |
| 974 | diesel::sql_query(if cfg!(feature = "sqlite") { |
| 975 | r#" |
| 976 | update |
| 977 | device |
| 978 | set |
| 979 | scheduler_run_after = ?3 |
| 980 | where |
| 981 | dev_eui in ( |
| 982 | select |
| 983 | d.dev_eui |
| 984 | from |
| 985 | device d |
| 986 | where |
| 987 | d.enabled_class in ('B', 'C') |
| 988 | and (d.scheduler_run_after is null or d.scheduler_run_after < ?2) |
| 989 | and d.is_disabled = FALSE |
| 990 | and exists ( |
| 991 | select |
| 992 | 1 |
| 993 | from |
| 994 | device_queue_item dq |
| 995 | where |
| 996 | dq.dev_eui = d.dev_eui |
| 997 | and not ( |
| 998 | -- pending queue-item with timeout_after in the future |
| 999 | (dq.is_pending = true and dq.timeout_after > ?2) |
| 1000 | ) |
| 1001 | ) |
| 1002 | order by d.dev_eui |
| 1003 | limit ?1 |
| 1004 | ) |
| 1005 | returning * |
| 1006 | "# |
| 1007 | } else { |
| 1008 | r#" |
| 1009 | update |
| 1010 | device |
| 1011 | set |