Feature: Session windows

docs/windowing.md

@@ -9,7 +9,9 @@ With windows, you can calculate such aggregations as:
 - Total of website visitors for every hour
 - The average speed of a vehicle over the last 10 minutes
 - Maximum temperature of a sensor observed over 30 second ranges
-- Give an user a reward after 10 succesful actions 
+- Give a user a reward after 10 successful actions
+- Track user activity sessions on a website
+- Detect fraud patterns in financial transactions 
 
 
 ## Types of Time in Streaming
@@ -203,14 +205,14 @@ Count-based Tumbling Windows slice incoming events into batch of a fixed size.
 For example, a tumbling window configured with a count of 4 will batch and aggregate message 1 to 4, then 5 to 8, 9 to 12 and so on. 
 
 ```
-Count       Tumbing Windows
+Count       Tumbling Windows
 [0, 3]  : ....
 [4, 7]  :     ....
 [8, 11] :         ....
 [12, 15] :            ....
 ```
 
-In a tumbing window each message is only assigned to a **single** interval.
+In a tumbling window each message is only assigned to a **single** interval.
 
 **Example**
 
@@ -356,7 +358,7 @@ Count       Hopping Windows
 [6, 11] :       ......
 ```
 
-In hopping windows each messages can be assigned to multiple windows because the windows overlap.
+In hopping windows each message can be assigned to multiple windows because the windows overlap.
 
 ## Time-based Sliding Windows
 Sliding windows are overlapping time-based windows that advance with each incoming message, rather than at fixed time intervals like hopping windows. They have a fixed 1 ms resolution and perform better and are less resource-intensive than hopping windows with a 1 ms step. Sliding windows do not produce redundant windows; every interval has a distinct aggregation.
@@ -438,7 +440,7 @@ sdf = (
 
 Sliding windows are overlapping windows that advance with each incoming message. They are equal to count-based hopping windows with a step of 1.
 
-For example a sliding window of 4 messagew will generate the followiwng windows:
+For example a sliding window of 4 messages will generate the following windows:
 
 ```
 Count       Sliding Windows
@@ -500,6 +502,229 @@ sdf = (
 
 ```
 
+## Session Windows
+
+Session windows group events that occur within a specified timeout period. Unlike fixed-time windows (tumbling, hopping, sliding), session windows have dynamic durations based on the actual timing of events. This makes them ideal for user activity tracking, fraud detection, and other event-driven scenarios.
+
+A session starts with the first event and extends each time a new event arrives within the timeout period. The session closes after the timeout period with no new events.
+
+Key characteristics of session windows:
+
+- **Dynamic boundaries**: Each session can have different start and end times based on actual events
+- **Activity-based**: Sessions extend automatically when events arrive within the timeout period  
+- **Event-driven closure**: Sessions close when no events arrive within the timeout period
+- **Grace period support**: Late events can still extend sessions if they arrive within the grace period
+
+### How Session Windows Work
+
+```
+Time:    0    5    10   15   20   25   30   35   40   45   50
+Events:  A         B              C    D              E
+
+Timeout: 10 seconds
+Grace:   2 seconds
+
+Session 1: [0, 20] - Events A, B (B extends the session from A)
+Session 2: [25, 40] - Events C, D (D extends the session from C)
+Session 3: [45, 55] - Event E (session will close at 55 if no more events)
+```
+
+In this example:
+- Event A starts Session 1 at time 0, session would timeout at time 10
+- Event B arrives at time 10, extending Session 1 to timeout at time 20
+- Event C arrives at time 25, starting Session 2 (too late for Session 1)
+- Event D arrives at time 30, extending Session 2 to timeout at time 40
+- Event E arrives at time 45, starting Session 3
+
+### Basic Session Window Example
+
+Imagine you want to track user activity sessions on a website, where a session continues as long as user actions occur within 30 minutes of each other:
+
+Input:
+```json
+{"user_action": "page_view", "page": "/home", "timestamp": 1000}
+{"user_action": "click", "element": "button", "timestamp": 800000}
+{"user_action": "page_view", "page": "/products", "timestamp": 1200000}
+{"user_action": "purchase", "amount": 50, "timestamp": 2000000}
+```
+
+Here's how to track user sessions using session windows:
+
+```python
+from datetime import timedelta
+from quixstreams import Application
+from quixstreams.dataframe.windows import Count, Collect
+
+app = Application(...)
+sdf = app.dataframe(...)
+
+sdf = (
+    # Define a session window with 30-minute timeout and 5-minute grace period
+    .session_window(
+        timeout_ms=timedelta(minutes=30),
+        grace_ms=timedelta(minutes=5)
+    )
+
+    # Count the number of actions in each session and collect all actions
+    .agg(
+        action_count=Count(),
+        actions=Collect("user_action")
+    )
+
+    # Emit results when sessions are complete
+    .final()
+)
+
+# Expected output (when session expires):
+# {
+#   "start": 1000,
+#   "end": 2000000,  # timestamp of last event
+#   "action_count": 4,
+#   "actions": ["page_view", "click", "page_view", "purchase"]
+# }
+```
+
+### Session Window with Current Mode
+
+For real-time monitoring, you can use `.current()` mode to get updates as the session progresses:
+
+Input:
+```json
+{"amount": 25, "timestamp": 1000}
+{"amount": 50, "timestamp": 5000} 
+{"amount": 50, "timestamp": 8000}
+```
+
+```python
+from datetime import timedelta
+from quixstreams import Application
+from quixstreams.dataframe.windows import Sum, Count
+
+app = Application(...)
+sdf = app.dataframe(...)
+
+sdf = (
+    # Define a session window with 10-second timeout
+    .session_window(timeout_ms=timedelta(seconds=10))
+
+    # Track total purchase amount and count in each session
+    .agg(
+        total_amount=Sum("amount"),
+        purchase_count=Count()
+    )
+
+    # Emit updates for each message (real-time session tracking)
+    .current()
+)
+
+# Output for each incoming event:
+# Event 1: {"start": 1000, "end": 1000, "total_amount": 25, "purchase_count": 1}
+# Event 2: {"start": 1000, "end": 5000, "total_amount": 75, "purchase_count": 2}
+# Event 3: {"start": 1000, "end": 8000, "total_amount": 125, "purchase_count": 3}
+```
+
+### Handling Late Events in Sessions
+
+Session windows support grace periods to handle out-of-order events:
+
+```python
+from datetime import timedelta
+from quixstreams import Application
+from quixstreams.dataframe.windows import Count
+
+def on_late_session_event(
+    value, key, timestamp_ms, late_by_ms, start, end, name, topic, partition, offset
+):
+    """Handle late events that couldn't extend any session"""
+    print(f"Late event for key {key}: {late_by_ms}ms late")
+    print(f"Event would have belonged to session [{start}, {end}]")
+    return False  # Suppress default logging
+
+app = Application(...)
+sdf = app.dataframe(...)
+
+sdf = (
+    # Session window with 5-minute timeout and 1-minute grace period
+    .session_window(
+        timeout_ms=timedelta(minutes=5),
+        grace_ms=timedelta(minutes=1),
+        on_late=on_late_session_event
+    )
+    .agg(event_count=Count())
+    .final()
+)
+```
+
+### Session Window Use Cases
+
+**1. User Activity Tracking**
+```python
+# Track user sessions on a website or app
+.session_window(timeout_ms=timedelta(minutes=30))
+.agg(
+    page_views=Count(),
+    unique_pages=Count("page_url", unique=True),
+    session_duration=Max("timestamp") - Min("timestamp")
+)
+```
+
+**2. Fraud Detection**
+```python
+# Detect suspicious transaction patterns
+.session_window(timeout_ms=timedelta(minutes=10))
+.agg(
+    transaction_count=Count(),
+    total_amount=Sum("amount"),
+    locations=Collect("location")
+)
+```
+
+**3. IoT Device Monitoring**
+```python
+# Monitor device activity sessions
+.session_window(timeout_ms=timedelta(hours=1))
+.agg(
+    readings_count=Count(),
+    avg_temperature=Mean("temperature"),
+    max_pressure=Max("pressure")
+)
+```
+
+**4. Gaming Analytics**
+```python
+# Track gaming sessions
+.session_window(timeout_ms=timedelta(minutes=20))
+.agg(
+    actions_performed=Count(),
+    points_earned=Sum("points"),
+    levels_completed=Count("level_completed")
+)
+```
+
+### Session Window Parameters
+
+- **`timeout_ms`**: The session timeout period. If no new events arrive within this period, the session will be closed. Can be specified as either an `int` (milliseconds) or a `timedelta` object.
+
+- **`grace_ms`**: The grace period for data arrival. Allows late-arriving data to be included in the session, even if it arrives after the session has theoretically timed out. Can be specified as either an `int` (milliseconds) or a `timedelta` object.
+
+- **`name`**: Optional unique identifier for the window. If not provided, it will be automatically generated based on the window's properties.
+
+- **`on_late`**: Optional callback to react to late records that cannot extend any existing session. Use this to customize logging or route late events to a dead-letter queue.
+
+### Session Window Behavior
+
+**Session Creation**: A new session starts when an event arrives and no existing session can accommodate it (i.e., all existing sessions have timed out).
+
+**Session Extension**: An existing session is extended when an event arrives within `timeout + grace_period` of the session's last activity.
+
+**Session Closure**: A session closes when the current time exceeds `last_event_time + timeout + grace_period`. The session end time in the output represents the timestamp of the last event in the session.
+
+**Out-of-Order Events**: When out-of-order events arrive within the grace period, they extend the session but do not change the end time if they are older than the current latest event. The end time always represents the timestamp of the chronologically latest event in the session.
+
+**Key Grouping**: Like all windows in Quix Streams, sessions are grouped by message key. Each key maintains its own independent sessions.
+
+**Event Time**: Sessions use event time (from Kafka message timestamps) rather than processing time.
+
 ## Lateness and Out-of-Order Processing
 When working with event time, some events may be processed later than they're supposed to.  
 Such events are called **"out-of-order"** because they violate the expected order of time in the data stream. 
@@ -540,7 +765,7 @@ The appropriate value for a grace period varies depending on the use case.
 ### Reacting on late events 
 !!! info New in v3.8.0
 
-To react on late records coming into time windows, you can pass the `on_late` callbacks to `.tumbling_window()`, `.hopping_window()` and `.sliding_window()` methods.
+To react on late records coming into time windows, you can pass the `on_late` callbacks to `.tumbling_window()`, `.hopping_window()`, `.sliding_window()`, and `.session_window()` methods.
 
 You can use this callback to customize the logging of such messages or to send them to some dead-letter queue, for example.
 
@@ -667,6 +892,8 @@ In this strategy, messages advance time and close only windows with the **same**
 
 If some message keys appear irregularly in the stream, the latest windows can remain unprocessed until the message with the same key is received.
 
+Session windows also support both closing strategies. With **key** strategy, sessions for each key close independently. With **partition** strategy, any message can advance time and close sessions for all keys in the partition.
+
 ```python
 from datetime import timedelta
 from quixstreams import Application
@@ -696,7 +923,7 @@ sdf = sdf.tumbling_window(timedelta(seconds=10)).agg(value=Sum()).final(closing_
 An alternative is to use the **partition** closing strategy.  
 In this strategy, messages advance time and close windows for the whole partition to which this key belongs.
 
-If messages aren't ordered accross keys some message can be skipped if the windows are already closed.
+If messages aren't ordered across keys some message can be skipped if the windows are already closed.
 
 ```python
 from datetime import timedelta
@@ -780,7 +1007,7 @@ described in [the "Updating Kafka Headers" section](./processing.md#updating-kaf
 
 Here are some general concepts about how windowed aggregations are implemented in Quix Streams:
 
-- Only time-based windows are supported. 
+- Time-based windows (tumbling, hopping, sliding, session) and count-based windows are supported.
 - Every window is grouped by the current Kafka message key.
 - Messages with `None` key will be ignored.
 - The minimal window unit is a **millisecond**. More fine-grained values (e.g. microseconds) will be rounded towards the closest millisecond number.
@@ -794,10 +1021,12 @@ window specification.
 
 The state store name is auto-generated by default using the following window attributes:
 
-- Window type: `"tumbling"` or `"hopping"`
-- Window parameters: `duration_ms` and `step_ms`
+- Window type: `"tumbling"`, `"hopping"`, `"sliding"`, or `"session"`
+- Window parameters: `duration_ms` and `step_ms` for time-based windows, `timeout_ms` for session windows
 
-E.g. a store name for a hopping window of 30 seconds with a 5 second step will be `hopping_window_30000_5000`.
+Examples:
+- A hopping window of 30 seconds with a 5 second step: `hopping_window_30000_5000`
+- A session window with 30 second timeout: `session_window_30000`
 
 ### Updating Window Definitions
 
@@ -807,8 +1036,8 @@ When you change the definition of the window (e.g. its size), the data in the st
 
 Quix Streams handles some of the situations, like:
 
-- Updating window type (e.g. from tumbling to hopping)
-- Updating window period or step 
+- Updating window type (e.g. from tumbling to hopping, from hopping to session)
+- Updating window period, step, or timeout
 - Adding/Removing/Updating an aggregation function (except `Reduce()`)
 
 Updating the window type and parameters will change the name of the underlying state store, and the new window definition will use a different one.