diff --git a/src/DesktopAnalytics/Analytics.cs b/src/DesktopAnalytics/Analytics.cs index 36fe51c..1e7c5cb 100644 --- a/src/DesktopAnalytics/Analytics.cs +++ b/src/DesktopAnalytics/Analytics.cs @@ -11,6 +11,7 @@ using System.Reflection; using System.Runtime.InteropServices; using System.Threading; +using System.Threading.Tasks; using System.Xml.Linq; using System.Xml.XPath; using JetBrains.Annotations; @@ -738,6 +739,19 @@ public void Dispose() _client?.ShutDown(); } + /// + /// Async alternative to for hosts that shut down asynchronously: + /// flushes/shuts down the underlying client without blocking a thread on any in-flight + /// delivery. Bounded just like Dispose -- an offline shutdown still returns promptly, with + /// undelivered events left on disk (Mixpanel) or in the transport library's own store + /// (Segment) for the next launch. Calling afterwards (e.g. from a + /// using block wrapping this object) is a harmless no-op. + /// + public Task ShutDownAsync() + { + return _client?.ShutDownAsync() ?? Task.CompletedTask; + } + /// /// Indicates whether we are tracking or not /// @@ -765,6 +779,34 @@ public static bool AllowTracking s_singleton.Initialize(initializationParameters); return; // Initialize sets s_allowTracking = true } + + // Already initialized and now re-enabled: re-arm any background flush loop that a + // prior PurgeQueuedEvents (consent revocation) paused. See offline-analytics.md. + try + { + s_singleton._client?.ResumeSending(); + } + catch (Exception e) + { + Debug.WriteLine("Analytics.AllowTracking: ResumeSending failed: " + e); + } + } + + if (!value) + { + // Consent revoked: purge any durable client's on-disk spool immediately (see + // offline-analytics.md, "Consent & lifecycle"). Guarded against a null singleton + // (AllowTracking can technically be set before an Analytics object is + // constructed) and a null client (the Segment/Mixpanel client is only assigned + // inside the Analytics constructor). + try + { + s_singleton?._client?.PurgeQueuedEvents(); + } + catch (Exception e) + { + Debug.WriteLine("Analytics.AllowTracking: PurgeQueuedEvents failed: " + e); + } } s_allowTracking = value; @@ -1084,5 +1126,15 @@ public static void FlushClient() { s_singleton._client?.Flush(); } + + /// + /// Async counterpart of : attempts delivery of anything pending + /// without blocking a thread on the network. Bounded -- returns promptly even while + /// offline, leaving undelivered events queued. + /// + public static Task FlushClientAsync() + { + return s_singleton._client?.FlushAsync() ?? Task.CompletedTask; + } } } diff --git a/src/DesktopAnalytics/AnalyticsEvent.cs b/src/DesktopAnalytics/AnalyticsEvent.cs new file mode 100644 index 0000000..f85c84e --- /dev/null +++ b/src/DesktopAnalytics/AnalyticsEvent.cs @@ -0,0 +1,83 @@ +using System; +using System.Text; +using Segment.Serialization; + +namespace DesktopAnalytics +{ + /// + /// A single spoolable analytics event in a neutral, JSON-serializable form. This is the + /// unit written to and read from the on-disk event spool (DiskQueue), independent of the + /// eventual transport (Mixpanel, etc). Instances are round-tripped as UTF-8 JSON bytes so + /// the spool stays portable and inspectable (see /). + /// + internal class AnalyticsEvent + { + public string AnalyticsId { get; set; } + public string EventName { get; set; } + + /// Event properties, e.g. Message/Stack Trace for exception reports. + public JsonObject Properties { get; set; } + + /// Mixpanel's $insert_id, stamped at enqueue time so at-least-once replay can + /// be safely deduplicated by the backend. + public string InsertId { get; set; } + + /// The original event time, stamped at enqueue time so a later replay is + /// back-dated correctly rather than reported at delivery time. + public DateTimeOffset Time { get; set; } + + public AnalyticsEvent() + { + Properties = new JsonObject(); + } + + /// + /// Creates a new event, stamping and . Both are + /// overridable so callers/tests can be deterministic; production code can omit them to get + /// a real GUID and the real wall-clock time. + /// + /// The stable per-user analytics id. + /// The event name. + /// Event properties. May be null, in which case an empty + /// property bag is used. + /// The $insert_id. Defaults to a fresh GUID. Never call + /// Guid.NewGuid() directly elsewhere in the stamping path -- pass it here so tests are + /// deterministic. + /// The event time. Defaults to DateTimeOffset.UtcNow. Never + /// call DateTime.Now/DateTimeOffset.UtcNow directly elsewhere in the stamping path -- + /// pass it here (e.g. from an injected ) so tests are + /// deterministic. + public static AnalyticsEvent Create( + string analyticsId, + string eventName, + JsonObject properties = null, + string insertId = null, + DateTimeOffset? time = null + ) + { + return new AnalyticsEvent + { + AnalyticsId = analyticsId, + EventName = eventName, + Properties = properties ?? new JsonObject(), + InsertId = insertId ?? Guid.NewGuid().ToString(), + Time = time ?? DateTimeOffset.UtcNow + }; + } + + /// Serializes this event as UTF-8 JSON bytes, suitable for enqueuing into the + /// on-disk spool. + public byte[] ToBytes() + { + var json = JsonUtility.ToJson(this, false); + return Encoding.UTF8.GetBytes(json); + } + + /// Deserializes an event previously produced by . + public static AnalyticsEvent FromBytes(byte[] bytes) + { + var json = Encoding.UTF8.GetString(bytes); + return JsonUtility.FromJson(json); + } + } +} diff --git a/src/DesktopAnalytics/DesktopAnalytics.csproj b/src/DesktopAnalytics/DesktopAnalytics.csproj index 379430c..fe12a54 100644 --- a/src/DesktopAnalytics/DesktopAnalytics.csproj +++ b/src/DesktopAnalytics/DesktopAnalytics.csproj @@ -17,10 +17,13 @@ ..\..\DesktopAnalytics.snk + All + + diff --git a/src/DesktopAnalytics/EventSpool.cs b/src/DesktopAnalytics/EventSpool.cs new file mode 100644 index 0000000..61f8454 --- /dev/null +++ b/src/DesktopAnalytics/EventSpool.cs @@ -0,0 +1,518 @@ +using System; +using System.Diagnostics; +using System.IO; +using System.Security.Cryptography; +using System.Text; +using System.Threading; +using System.Threading.Tasks; +using DiskQueue; + +namespace DesktopAnalytics +{ + /// + /// The outcome of attempting to deliver one spooled event, as reported by the caller of + /// . This drives whether the event is removed from the + /// spool (/) or left in place for a later + /// retry (). + /// + internal enum SendResult + { + /// The event was confirmed delivered (e.g. HTTP 2xx). Remove it from the spool. + Delivered, + + /// A transient failure (connection error, timeout, 5xx, 429). Leave the event in + /// the spool for a later retry. + RetryableFailure, + + /// A non-retryable failure (e.g. a 4xx that will never succeed). Remove the event + /// from the spool anyway so a single bad event cannot wedge the spool forever. + PoisonDrop + } + + /// + /// Durable, bounded, on-disk store of pending s, backed by + /// DiskQueue. Events are enqueued as opaque UTF-8 JSON bytes (see + /// ) so the spool stays portable and inspectable. + /// + /// + /// Every public method here swallows disk/IO/serialization exceptions internally (logging via + /// ) and returns gracefully -- analytics code must never + /// throw into the host application. The one exception is the constructor: failing to acquire + /// the underlying DiskQueue's cross-process exclusive lock is a startup-time condition the + /// caller needs to know about, so it is allowed to propagate. + /// + internal class EventSpool : IDisposable + { + // How long we wait to acquire DiskQueue's cross-process exclusive lock when opening the + // spool. Kept short: if another process is holding the lock for this long, something is + // wrong, and we'd rather fail fast than hang application startup/shutdown. + private static readonly TimeSpan s_lockWaitTimeout = TimeSpan.FromSeconds(2); + + // Number of leading bytes of the SHA-256 hash of the API key used to build the spool + // directory name. Just needs to be stable and distinguish keys from each other -- it is + // not a security boundary -- so a short prefix is plenty. + private const int kApiKeyHashBytes = 8; + + private readonly int _maxItems; + private readonly int _maxItemBytes; + private readonly long _maxSpoolBytes; + // Logical bytes of the LIVE entries (disk file sizes would overcount consumed-but-untrimmed + // ones). Guarded by _sync; measured at open, then maintained incrementally. + private long _spoolBytes; + private readonly string _spoolDirectory; + // Not readonly: Purge replaces the queue wholesale (see its remarks). Always accessed + // under _sync. Null only if a Purge-time reopen failed; every public method already + // catches and logs whatever follows from that. + private IPersistentQueue _queue; + // Serializes ALL access to _queue, exactly like the lock(_syncRoot) it replaced -- a + // SemaphoreSlim because ProcessBatchAsync must hold it across awaits of the send callback, + // which a monitor lock cannot. Deliberately never disposed: it holds no unmanaged + // resources unless AvailableWaitHandle is touched (it isn't), and disposing it would turn + // a benign use-after-Dispose race into ObjectDisposedException noise. + private readonly SemaphoreSlim _sync = new SemaphoreSlim(1, 1); + private bool _disposed; + + /// + /// Opens (creating if necessary) the on-disk event spool rooted at + /// , taking DiskQueue's cross-process exclusive lock for + /// the lifetime of this object. + /// + /// Directory to hold the spool's files. Callers normally get + /// this from ; tests typically inject a temp directory. + /// The maximum number of events retained. Enqueuing beyond this + /// drops the oldest event(s) first. + /// The maximum serialized size of a single event; anything + /// larger is refused at enqueue time ( returns false) rather than + /// spooled. Defaults to unbounded; passes the transport's + /// real per-event limit. + /// The maximum total serialized size of all retained events; + /// enqueuing beyond this drops the oldest event(s) first, exactly like + /// . Bounds both the disk footprint and the total upload + /// liability of an accumulated backlog. Defaults to unbounded. + /// Propagates whatever DiskQueue throws if the lock cannot be + /// acquired within the internal wait timeout (e.g. another process/instance already has + /// this spool open) or if the directory cannot be created/opened. This is deliberately not + /// swallowed: it is a startup-time failure the caller needs to know about. + public EventSpool(string spoolDirectory, int maxItems, int maxItemBytes = int.MaxValue, + long maxSpoolBytes = long.MaxValue) + { + if (maxItems < 0) + throw new ArgumentOutOfRangeException(nameof(maxItems)); + if (maxItemBytes <= 0) + throw new ArgumentOutOfRangeException(nameof(maxItemBytes)); + if (maxSpoolBytes <= 0) + throw new ArgumentOutOfRangeException(nameof(maxSpoolBytes)); + + _maxItems = maxItems; + _maxItemBytes = maxItemBytes; + _maxSpoolBytes = maxSpoolBytes; + _spoolDirectory = spoolDirectory; + _queue = PersistentQueue.WaitFor(spoolDirectory, s_lockWaitTimeout); + _spoolBytes = MeasureExistingSpoolBytes(); + } + + // Sums the live entries left by a previous session (dequeue-all in a session disposed + // without Flush = pure measurement; DiskQueue rolls it back). Failure => 0: the byte cap + // is then under-enforced until the backlog drains, erring toward keeping events. + private long MeasureExistingSpoolBytes() + { + try + { + long total = 0; + using (var session = _queue.OpenSession()) + { + while (true) + { + var bytes = session.Dequeue(); + if (bytes == null) + break; + total += bytes.Length; + } + } + return total; + } + catch (Exception e) + { + Debug.WriteLine("EventSpool: failed to measure existing spool bytes, assuming 0: " + e); + return 0; + } + } + + /// + /// Computes the per-user, per-API-key spool directory: + /// %LocalAppData%\SIL\DesktopAnalytics\spool\<short hash of apiKey>. The API key is + /// hashed (never embedded raw) so that, e.g., DEBUG and RELEASE builds of an app -- which + /// use different keys -- get separate spools, without exposing the key via the file system. + /// + public static string GetDefaultSpoolPath(string apiKey) + { + var root = Path.Combine( + Environment.GetFolderPath(Environment.SpecialFolder.LocalApplicationData), + "SIL", "DesktopAnalytics", "spool"); + return Path.Combine(root, HashApiKey(apiKey)); + } + + private static string HashApiKey(string apiKey) + { + using (var sha = SHA256.Create()) + { + var hash = sha.ComputeHash(Encoding.UTF8.GetBytes(apiKey ?? string.Empty)); + var builder = new StringBuilder(kApiKeyHashBytes * 2); + for (var i = 0; i < kApiKeyHashBytes; i++) + builder.Append(hash[i].ToString("x2")); + return builder.ToString(); + } + } + + /// + /// An approximate count of events currently in the spool. "Approximate" because DiskQueue + /// tracks this as an estimate that is cheap to read; it is accurate enough for bounding and + /// diagnostics but should not be treated as a hard guarantee under concurrent access. + /// + public int ApproximateCount + { + get + { + try + { + _sync.Wait(); + try + { + return _queue.EstimatedCountOfItemsInQueue; + } + finally + { + _sync.Release(); + } + } + catch (Exception e) + { + Debug.WriteLine("EventSpool.ApproximateCount: failed to read count: " + e); + return 0; + } + } + } + + /// + /// Serializes and enqueues , then enforces the configured item cap by + /// dropping the oldest event(s) if necessary. Never throws: any disk/IO/serialization + /// failure is logged and swallowed, and the event is simply lost rather than crashing the + /// host. + /// + /// True if the event was durably enqueued; false if it was dropped (null, + /// unserializable, larger than the per-event byte cap, or a disk failure). Callers use + /// this to count the drop -- see . + public bool Enqueue(AnalyticsEvent evt) + { + if (evt == null) + return false; + + try + { + byte[] bytes; + try + { + bytes = evt.ToBytes(); + } + catch (Exception e) + { + Debug.WriteLine("EventSpool.Enqueue: failed to serialize event, dropping: " + e); + return false; + } + + if (bytes.Length > _maxItemBytes) + { + // Refused up front rather than spooled: an event over the transport's per-event + // limit can never be delivered, and if it is large enough to time out the HTTP + // request it would be classified as a RETRYABLE failure (a timeout is + // indistinguishable from being offline) and wedge the head of the queue forever. + Debug.WriteLine("EventSpool.Enqueue: event of " + bytes.Length + + " bytes exceeds the " + _maxItemBytes + "-byte cap, dropping: " + evt.EventName); + return false; + } + + _sync.Wait(); + try + { + using (var session = _queue.OpenSession()) + { + session.Enqueue(bytes); + session.Flush(); + } + + _spoolBytes += bytes.Length; + EnforceCapLocked(); + } + finally + { + _sync.Release(); + } + + return true; + } + catch (Exception e) + { + Debug.WriteLine("EventSpool.Enqueue failed: " + e); + return false; + } + } + + // Caller must hold _sync. Drops the oldest item(s) one at a time until the queue is at or + // under BOTH caps (_maxItems and _maxSpoolBytes). Each iteration removes exactly one item + // (or gives up on a read/dequeue failure), so this always terminates -- including the + // degenerate case of _maxItems == 0, where even the item just enqueued is dropped. + private void EnforceCapLocked() + { + while (true) + { + int count; + try + { + count = _queue.EstimatedCountOfItemsInQueue; + } + catch (Exception e) + { + Debug.WriteLine("EventSpool.EnforceCap: failed to read count, giving up: " + e); + return; + } + + if (count <= _maxItems && _spoolBytes <= _maxSpoolBytes) + return; + + if (count <= 0) + { + // Nothing left to drop; if the byte estimate says otherwise it has drifted -- + // resync it to the truth (an empty queue holds zero bytes). + _spoolBytes = 0; + return; + } + + try + { + using (var session = _queue.OpenSession()) + { + var oldest = session.Dequeue(); + if (oldest == null) + { + _spoolBytes = 0; // See the count <= 0 case above. + return; + } + session.Flush(); + _spoolBytes = Math.Max(0, _spoolBytes - oldest.Length); + } + } + catch (Exception e) + { + Debug.WriteLine("EventSpool.EnforceCap: failed to drop oldest item, giving up: " + e); + return; + } + } + } + + // Test seam: the spool's current logical byte total (the value the byte cap is enforced + // against), primarily so tests can prove it survives a dispose/reopen. + internal long ApproximateBytes + { + get + { + _sync.Wait(); + try + { + return _spoolBytes; + } + finally + { + _sync.Release(); + } + } + } + + /// + /// Attempts to deliver up to spooled events, oldest first, in + /// a single DiskQueue session/transaction. For each dequeued event, + /// decides its fate: + /// + /// or : the + /// dequeue is committed (session.Flush()) and processing continues with the next + /// event. + /// : the dequeue is not committed, + /// and processing stops -- disposing the session without flushing rolls that event (and + /// only that event) back into the spool for a later retry. + /// + /// If throws, it is treated exactly like + /// : the exception is swallowed, nothing is + /// flushed, and processing stops. This is the crash-window path that guarantees + /// at-least-once delivery -- a sender that delivered the event but crashed/threw before + /// this method could flush will see the same event again on the next call. + /// + /// Maximum number of events attempted in this call. + /// Decides each event's fate; see the summary above. + /// Soft byte budget for this call, bounding each drain burst on a + /// slow/metered connection. Checked AFTER each event, so a single event over the whole + /// budget still goes -- the budget can never starve the queue. + public async Task ProcessBatchAsync(int maxItems, Func> send, + long maxBytes = long.MaxValue) + { + if (send == null || maxItems <= 0 || maxBytes <= 0) + return; + + try + { + await _sync.WaitAsync().ConfigureAwait(false); + try + { + long sentBytes = 0; + using (var session = _queue.OpenSession()) + { + for (var i = 0; i < maxItems; i++) + { + byte[] bytes; + try + { + bytes = session.Dequeue(); + } + catch (Exception e) + { + Debug.WriteLine("EventSpool.ProcessBatch: dequeue failed: " + e); + return; + } + + if (bytes == null) + return; // Spool is empty. + + AnalyticsEvent evt; + try + { + evt = AnalyticsEvent.FromBytes(bytes); + } + catch (Exception e) + { + // Unreadable entry (corrupt, or from an incompatible schema version). + // Treat it like a poison message rather than wedging the spool. + Debug.WriteLine( + "EventSpool.ProcessBatch: failed to deserialize event, dropping: " + e); + session.Flush(); + _spoolBytes = Math.Max(0, _spoolBytes - bytes.Length); + continue; + } + + SendResult result; + try + { + result = await send(evt).ConfigureAwait(false); + } + catch (Exception e) + { + Debug.WriteLine( + "EventSpool.ProcessBatch: send threw, leaving event for retry: " + e); + return; // Do not flush -- this dequeue rolls back on session dispose. + } + + if (result == SendResult.Delivered || result == SendResult.PoisonDrop) + { + session.Flush(); + _spoolBytes = Math.Max(0, _spoolBytes - bytes.Length); + sentBytes += bytes.Length; + if (sentBytes >= maxBytes) + return; // Byte budget for this call is spent. + continue; + } + + // RetryableFailure: stop without flushing so this event rolls back. + return; + } + } + } + finally + { + _sync.Release(); + } + } + catch (Exception e) + { + Debug.WriteLine("EventSpool.ProcessBatch failed: " + e); + } + } + + /// + /// Empties the spool entirely (used on consent revocation), removing the event data from + /// disk -- not merely marking it consumed. Never throws. + /// + /// + /// Implemented as dispose + delete-the-directory + reopen rather than either alternative, + /// both tried empirically (2026-07, DiskQueue 1.7.2): a dequeue-and-flush loop leaves the + /// purged events' bytes sitting in the data files (DiskQueue does not trim consumed + /// entries), which is not acceptable for a CONSENT purge; and DiskQueue's own + /// HardDelete(reset: true) leaves the live instance's + /// EstimatedCountOfItemsInQueue stale, so the spool keeps reporting the purged + /// events as present -- consistent with its "not thread safe ... or safe in any other + /// way" warning. Between the dispose and the reopen the cross-process exclusive lock is + /// briefly released; if another process steals it in that window the reopen fails, this + /// spool degrades to a no-op (every method here already tolerates that), and the purge + /// itself has still succeeded -- the data is gone, which is the property that matters. + /// + public void Purge() + { + try + { + _sync.Wait(); + try + { + if (_disposed) + return; + + try + { + _queue?.Dispose(); + } + finally + { + _queue = null; + } + + if (Directory.Exists(_spoolDirectory)) + Directory.Delete(_spoolDirectory, true); + + _queue = PersistentQueue.WaitFor(_spoolDirectory, s_lockWaitTimeout); + _spoolBytes = 0; + } + finally + { + _sync.Release(); + } + } + catch (Exception e) + { + Debug.WriteLine("EventSpool.Purge failed: " + e); + } + } + + /// + /// Releases the underlying DiskQueue's cross-process exclusive lock. Safe to call more + /// than once. + /// + public void Dispose() + { + if (_disposed) + return; + + try + { + _sync.Wait(); + try + { + _disposed = true; + _queue?.Dispose(); + } + finally + { + _sync.Release(); + } + } + catch (Exception e) + { + Debug.WriteLine("EventSpool.Dispose failed: " + e); + } + } + } +} diff --git a/src/DesktopAnalytics/IClient.cs b/src/DesktopAnalytics/IClient.cs index 1923115..bdf1cc6 100644 --- a/src/DesktopAnalytics/IClient.cs +++ b/src/DesktopAnalytics/IClient.cs @@ -1,4 +1,5 @@ -using Segment.Serialization; +using System.Threading.Tasks; +using Segment.Serialization; namespace DesktopAnalytics { @@ -9,6 +10,38 @@ internal interface IClient void Identify(string analyticsId, JsonObject traits, JsonObject options); void Track(string defaultIdForAnalytics, string eventName, JsonObject properties); void Flush(); + + /// + /// Async counterpart of . Implementations whose shutdown has nothing + /// awaitable (see ) may complete synchronously; implementations + /// with real async delivery work (see ) must never block a + /// thread waiting on it. Like every member here, must not throw. + /// + Task ShutDownAsync(); + + /// + /// Async counterpart of . Same contract notes as + /// . + /// + Task FlushAsync(); + Statistics Statistics { get; } + + /// + /// Called when the user revokes tracking consent ( + /// transitioning to false). Implementations that spool events on disk (see + /// ) must purge that spool immediately; implementations without a + /// local spool (see ) can no-op. + /// + void PurgeQueuedEvents(); + + /// + /// Called when the user grants tracking consent again ( + /// transitioning back to true) on an already-initialized client, after a prior + /// paused it. Implementations with a background flush loop (see + /// ) must re-arm it; implementations without one (see + /// ) can no-op. + /// + void ResumeSending(); } } \ No newline at end of file diff --git a/src/DesktopAnalytics/IEventSender.cs b/src/DesktopAnalytics/IEventSender.cs new file mode 100644 index 0000000..421fc67 --- /dev/null +++ b/src/DesktopAnalytics/IEventSender.cs @@ -0,0 +1,27 @@ +using System.Threading; +using System.Threading.Tasks; + +namespace DesktopAnalytics +{ + /// + /// The one seam between the durable and the network. Implementations + /// deliver a single and classify the outcome via + /// (defined alongside ) so the flush loop knows whether to remove the event + /// from the spool, leave it for retry, or drop it as a poison message. + /// + /// + /// Implementations must never throw (synchronously or as a faulted task): the flush loop (via the + /// Polly-wrapped call in ) treats an exception the same as + /// , but a well-behaved sender should catch its own + /// network/timeout exceptions and return directly. + /// + internal interface IEventSender + { + /// Allows a bounded caller (see + /// 's BoundedDrain, used by Flush/ShutDown) to abandon a send + /// that is taking too long -- e.g. a server that accepts the connection but never responds + /// (a "black hole"). Implementations should treat cancellation like any other transient + /// failure and return rather than throwing. + Task SendAsync(AnalyticsEvent evt, CancellationToken cancellationToken = default); + } +} diff --git a/src/DesktopAnalytics/MixpanelClient.cs b/src/DesktopAnalytics/MixpanelClient.cs index 2ce3493..1f8e7b7 100644 --- a/src/DesktopAnalytics/MixpanelClient.cs +++ b/src/DesktopAnalytics/MixpanelClient.cs @@ -1,58 +1,694 @@ -using System; -using System.Collections.Generic; -using System.Linq; +using System; +using System.Diagnostics; +using System.Net.NetworkInformation; using System.Threading; using System.Threading.Tasks; +using Polly; +using Polly.CircuitBreaker; +using Polly.Retry; using Segment.Serialization; namespace DesktopAnalytics { + /// + /// Durable Mixpanel client: events are scrubbed, stamped, and written to an on-disk + /// immediately (never lost to being offline), and a background flush + /// loop drains the spool through a Polly-wrapped whenever it can. See + /// offline-analytics.md for the design this implements. + /// + /// + /// Every public member here swallows exceptions and returns/no-ops rather than throwing -- + /// analytics must never crash the host application (see offline-analytics.md, "Never crash the + /// host"). The one intentionally-not-swallowed failure is a bad + /// constructor call inside , which is itself caught here so it cannot + /// escape to the caller either; it just leaves this client running with no spool (Track/Drain + /// become no-ops) rather than durable. + /// internal class MixpanelClient : IClient { - private Mixpanel.MixpanelClient _client; + // Open questions in offline-analytics.md: exact cap/batch/cadence values are not locked yet. + // These are reasonable starting defaults, not requirements baked in elsewhere. + private const int kDefaultMaxSpoolItems = 5000; + private const int kDefaultBatchSize = 20; + private const int kDefaultFlushIntervalSeconds = 30; - private readonly List> _tasks = new List>(); + // Mixpanel's documented per-event limit: 1MB of uncompressed JSON + // (https://docs.mixpanel.com/reference/import-events, "Common Issues"). An event over this + // can never be delivered, so it is refused at enqueue time (counted in Statistics.Failed) + // instead of spooled -- if it merely got rejected we'd catch it as a PoisonDrop, but one + // large enough to time out the HTTP request would classify as RETRYABLE (a timeout is + // indistinguishable from being offline) and wedge the head of the queue forever. + private const int kMaxSpooledEventBytes = 1024 * 1024; - public void Initialize(string apiSecret, string host = null, int _flushAt = -1, int _flushInterval = -1) + // Bandwidth courtesy on slow/metered connections: a soft byte budget per flush tick, and a + // total-backlog cap (drop-oldest) bounding disk footprint and upload liability. See the + // "Constrained bandwidth" section of the PR #43 description for the pacing rationale. + private const long kMaxBytesPerDrainTick = 256 * 1024; + private const long kDefaultMaxSpoolBytes = 20L * 1024 * 1024; + + // On (re)start, attempt the first drain soon rather than waiting a full interval, so events + // spooled during a PREVIOUS (offline) session go out shortly after launch instead of ~30s later. + private const int kInitialFlushDelaySeconds = 3; + // When we get an opportunistic OS signal that connectivity may have returned (a network address + // change), reschedule the next drain this soon instead of waiting for the next poll tick. + private const int kReconnectFlushDelaySeconds = 1; + + // Bound on Flush()/ShutDown(): an offline shutdown must return fast rather than hang the + // host's exit. Two independent bounds (wall-clock + attempt count) so either one alone + // failing to trip (e.g. a TimeProvider that never advances in a test) still terminates. + private static readonly TimeSpan s_boundedDrainDuration = TimeSpan.FromSeconds(5); + private const int kBoundedDrainMaxAttempts = 20; + + private EventSpool _spool; + private IEventSender _sender; + private TimeProvider _timeProvider = TimeProvider.System; + private ResiliencePipeline _pipeline = ResiliencePipeline.Empty; + private Mixpanel.MixpanelClient _identifyClient; + private Timer _flushTimer; + private int _batchSize = kDefaultBatchSize; + // Normalized once in Initialize (clamped to >= 1s there); everywhere else just reads it. + private TimeSpan _flushInterval = TimeSpan.FromSeconds(kDefaultFlushIntervalSeconds); + // Set by Initialize/InitializeForTest. Track() throws if this is still false -- calling + // Track on a never-initialized client is a programming error in the host, matching the + // pre-durability behavior (the old client dereferenced a null field there). + private bool _initialized; + // Paused == consent revoked (see PurgeQueuedEvents): the poll timer is stopped and opportunistic + // network-change kicks are ignored until ResumeSending re-enables them. + private volatile bool _paused; + + private int _submitted; + private int _succeeded; + private int _failed; + + // Guards only against overlapping TIMER ticks: if a previous tick's drain is still running + // when the next tick fires, the new tick is skipped. It does NOT serialize ticks against + // Flush/ShutDown/DrainOnceAsync -- those may run concurrently with a tick, which is safe + // because all spool access is serialized inside EventSpool (its semaphore); this flag just + // keeps a slow drain from stacking up redundant timer callbacks behind it. + private int _timerDraining; + + /// + /// Production initializer (via ). Builds a real on-disk spool keyed by + /// , a real HTTP-posting , and a + /// Polly retry+circuit-breaker pipeline, then starts the background flush timer. + /// + /// Not supported by the Mixpanel client; passing a non-empty value throws + /// (this is the long-standing contract — only + /// honors a host). + public void Initialize(string apiSecret, string host = null, int flushAt = -1, int flushInterval = -1) { - // currently only the SegmentClient uses the host parameter - if (host != null) + // Preserve the original contract: unlike SegmentClient, the Mixpanel client does not support + // pointing at a different host. Validated up front, OUTSIDE the catch below, so a + // misconfiguration surfaces to the caller exactly as it always has rather than being swallowed. + if (!string.IsNullOrEmpty(host)) + throw new ArgumentException("MixpanelClient does not currently support a host parameter", nameof(host)); + + _initialized = true; + + try { - throw new ArgumentException("MixpanelClient does not currently support a host parameter"); + _batchSize = flushAt > 0 ? flushAt : kDefaultBatchSize; + // Clamp to >= 1s once, here, so every later consumer can use the value as-is. + _flushInterval = TimeSpan.FromSeconds( + Math.Max(1, flushInterval > 0 ? flushInterval : kDefaultFlushIntervalSeconds)); + + _timeProvider = TimeProvider.System; + _spool = new EventSpool(EventSpool.GetDefaultSpoolPath(apiSecret), kDefaultMaxSpoolItems, + kMaxSpooledEventBytes, kDefaultMaxSpoolBytes); + _sender = new MixpanelEventSender(apiSecret); + _pipeline = BuildDefaultPipeline(_timeProvider); + + try + { + // Best-effort only; used solely for the (unspooled) Identify call. + _identifyClient = new Mixpanel.MixpanelClient(apiSecret); + } + catch (Exception e) + { + Debug.WriteLine("MixpanelClient.Initialize: failed to create identify client: " + e); + } + + StartTimer(); + SubscribeToNetworkChanges(); + } + catch (Exception e) + { + Debug.WriteLine("MixpanelClient.Initialize failed: " + e); } - _client = new Mixpanel.MixpanelClient(apiSecret); } - public void ShutDown() + /// + /// Test-only initializer: injects every non-deterministic dependency directly (the spool, the + /// sender, the clock, and the resilience pipeline) and deliberately does NOT start the + /// background timer -- tests pump manually instead of racing a real + /// timer thread. See offline-analytics.md, "Design-for-test seams". + /// + /// If null, defaults to + /// (no retry, no circuit breaker) so that, by default, one call maps + /// to exactly one call per spooled event -- the simplest, + /// most deterministic shape for most tests. Tests that specifically want to exercise Polly + /// retry/circuit-breaker behavior pass their own pipeline (see + /// ). + internal void InitializeForTest( + EventSpool spool, + IEventSender sender, + TimeProvider timeProvider = null, + ResiliencePipeline pipeline = null, + int batchSize = kDefaultBatchSize) + { + _initialized = true; + _spool = spool; + _sender = sender; + _timeProvider = timeProvider ?? TimeProvider.System; + _pipeline = pipeline ?? ResiliencePipeline.Empty; + _batchSize = batchSize; + } + + /// + /// Builds the production Polly pipeline: a small bounded retry (exponential backoff + + /// jitter) for quick transient blips, wrapped in a circuit breaker so a sustained outage backs + /// off instead of hammering the network on every flush tick. Exposed (internal, static) so + /// tests that specifically want to exercise real Polly retry/circuit-breaker behavior -- + /// rather than the zero-strategy default used by -- can build + /// one with a deterministic (e.g. a zero-delay variant; see + /// MixpanelClientTests for the rationale). + /// + internal static ResiliencePipeline BuildDefaultPipeline( + TimeProvider timeProvider, + int maxRetryAttempts = 2, + TimeSpan? retryDelay = null) + { + bool ShouldHandleOutcome(Outcome outcome) => + outcome.Exception != null || outcome.Result == SendResult.RetryableFailure; + + var retryOptions = new RetryStrategyOptions + { + ShouldHandle = args => new ValueTask(ShouldHandleOutcome(args.Outcome)), + MaxRetryAttempts = maxRetryAttempts, + Delay = retryDelay ?? TimeSpan.FromMilliseconds(250), + BackoffType = DelayBackoffType.Exponential, + UseJitter = true + }; + + var breakerOptions = new CircuitBreakerStrategyOptions + { + ShouldHandle = args => new ValueTask(ShouldHandleOutcome(args.Outcome)), + FailureRatio = 0.5, + // Matches attempts-per-tick: a fully failing drain tick produces exactly 3 + // outcomes through this pipeline (1 initial attempt + MaxRetryAttempts=2 retries), + // all within the same instant, so all 3 always land in one SamplingDuration window. + // With this at 4 (the previous value), a single tick could never reach the minimum + // throughput -- and successive ticks are kDefaultFlushIntervalSeconds (30s) apart, + // far outside the 10s window -- so the breaker could never open at all during a + // sustained outage. 3 makes one bad tick enough to trip it. + MinimumThroughput = 3, + SamplingDuration = TimeSpan.FromSeconds(10), + BreakDuration = TimeSpan.FromSeconds(5) + }; + + var builder = new ResiliencePipelineBuilder + { + TimeProvider = timeProvider ?? TimeProvider.System + }; + return builder + .AddRetry(retryOptions) + .AddCircuitBreaker(breakerOptions) + .Build(); + } + + private void StartTimer() { - var totalWait = 0; - while (_tasks.Any(t => !t.IsCompleted)) + try + { + // Fire the first drain soon after launch (bounded by the interval) so events left in the + // spool by a previous offline session are attempted promptly, not a full interval later. + var initialDelay = TimeSpan.FromTicks( + Math.Min(TimeSpan.FromSeconds(kInitialFlushDelaySeconds).Ticks, _flushInterval.Ticks)); + _paused = false; + // Fire-and-forget by design: OnTimerTickAsync catches everything internally, so the + // discarded task can never fault unobserved. + _flushTimer = new Timer(_ => _ = OnTimerTickAsync(), null, initialDelay, _flushInterval); + } + catch (Exception e) + { + Debug.WriteLine("MixpanelClient.StartTimer failed: " + e); + } + } + + private void SubscribeToNetworkChanges() + { + try + { + NetworkChange.NetworkAddressChanged += OnNetworkAddressChanged; + } + catch (Exception e) + { + // Not fatal: the periodic poll timer is the RELIABLE delivery mechanism. This subscription + // is only an opportunistic accelerator, and some platforms may not support it. + Debug.WriteLine("MixpanelClient.SubscribeToNetworkChanges failed: " + e); + } + } + + private void UnsubscribeFromNetworkChanges() + { + try + { + NetworkChange.NetworkAddressChanged -= OnNetworkAddressChanged; + } + catch (Exception e) { - if (totalWait > 7500) - break; - // REVIEW: I modeled this waiting off the segment client, there might be - // better metrics for Mixpanel, but I'm in a rush - totalWait += 500; - Thread.Sleep(500); + Debug.WriteLine("MixpanelClient.UnsubscribeFromNetworkChanges failed: " + e); } } + // Opportunistic accelerator: when the OS reports a network address change (e.g. Wi-Fi reconnecting + // after being offline), reschedule the next drain to happen almost immediately instead of waiting + // out the remainder of the poll interval. The periodic timer remains the guarantee -- this only + // improves reconnect latency. Ignored while paused (consent revoked). Never throws. + private void OnNetworkAddressChanged(object sender, EventArgs e) + { + try + { + if (_paused) + return; + if (!NetworkInterface.GetIsNetworkAvailable()) + return; + _flushTimer?.Change( + TimeSpan.FromSeconds(kReconnectFlushDelaySeconds), _flushInterval); + } + catch (Exception ex) + { + Debug.WriteLine("MixpanelClient.OnNetworkAddressChanged failed: " + ex); + } + } + + // Starts on a ThreadPool thread (System.Threading.Timer) and continues on the pool after + // awaits (ConfigureAwait(false) throughout the drain path) -- never the UI thread. + private async Task OnTimerTickAsync() + { + if (Interlocked.CompareExchange(ref _timerDraining, 1, 0) != 0) + return; // A previous tick is still draining; skip. + + try + { + await DrainOnceAsync().ConfigureAwait(false); + } + catch (Exception e) + { + Debug.WriteLine("MixpanelClient: background drain failed: " + e); + } + finally + { + Interlocked.Exchange(ref _timerDraining, 0); + } + } + + /// + /// One-shot attempt to deliver up to one batch of spooled events. This is exactly what the + /// background timer awaits on each tick; tests await it directly to pump the flush loop + /// deterministically instead of racing a real timer (see offline-analytics.md, "Flush-loop + /// scheduling" in the design-for-test-seams table). Always goes through the full Polly + /// pipeline (retry + circuit breaker) with no cancellation -- + /// is the bounded/cancellable variant used by Flush/ShutDown. + /// + internal Task DrainOnceAsync() + { + return DrainOnceCoreAsync(CancellationToken.None, usePipeline: true); + } + + // Shared implementation behind DrainOnceAsync() and BoundedDrainAsync(). usePipeline is false + // only for BoundedDrainAsync's bounded attempts (see its comment for why retries are skipped + // there). + private async Task DrainOnceCoreAsync(CancellationToken cancellationToken, bool usePipeline) + { + try + { + if (_spool == null) + return; + await _spool.ProcessBatchAsync(_batchSize, + evt => SendOneEventAsync(evt, cancellationToken, usePipeline), + kMaxBytesPerDrainTick) + .ConfigureAwait(false); + } + catch (Exception e) + { + Debug.WriteLine("MixpanelClient.DrainOnce failed: " + e); + } + } + + // The callback EventSpool.ProcessBatchAsync awaits per dequeued event. Must never throw or + // fault (see EventSpool.ProcessBatchAsync's contract: a thrown exception is treated as a + // RetryableFailure anyway, but returning it directly avoids relying on that fallback). + private async Task SendOneEventAsync(AnalyticsEvent evt, CancellationToken cancellationToken, + bool usePipeline) + { + try + { + SendResult result; + try + { + result = usePipeline + ? await _pipeline.ExecuteAsync( + async ct => await _sender.SendAsync(evt, ct).ConfigureAwait(false), + cancellationToken) + .ConfigureAwait(false) + : await _sender.SendAsync(evt, cancellationToken).ConfigureAwait(false); + } + catch (BrokenCircuitException e) + { + // Circuit is open from a sustained outage -- back off without even trying the + // network this round. Leave the event for a later retry. + Debug.WriteLine("MixpanelClient: circuit breaker open, leaving event for retry: " + e); + return SendResult.RetryableFailure; + } + catch (OperationCanceledException e) + { + // BoundedDrain's per-attempt deadline (see its comment) fired before the sender + // finished -- treat exactly like any other transient failure so the event stays + // spooled for the next drain, rather than letting this hang or throw into the host. + Debug.WriteLine( + "MixpanelClient: send canceled (bounded drain deadline), leaving event for retry: " + e); + return SendResult.RetryableFailure; + } + + switch (result) + { + case SendResult.Delivered: + Interlocked.Increment(ref _succeeded); + break; + case SendResult.PoisonDrop: + Interlocked.Increment(ref _failed); + break; + } + + return result; + } + catch (Exception e) + { + // The sender (or the pipeline itself) threw and retries -- if any -- were exhausted. + // Never let that propagate: leave the event in the spool for the next drain. + Debug.WriteLine("MixpanelClient: send pipeline threw unexpectedly, leaving event for retry: " + + e); + return SendResult.RetryableFailure; + } + } + + /// + /// Best-effort, non-spooled identify call. Deliberately out of scope for the durability work + /// here (see offline-analytics.md): traits are small, low-value to replay, and bundling them + /// into the same at-least-once spool as usage/exception events would complicate dedup for + /// little benefit. Never throws; failures are logged and otherwise ignored. + /// public void Identify(string analyticsId, JsonObject traits, JsonObject options) { - _tasks.Add(_client.PeopleSetAsync(analyticsId, traits)); + try + { + if (_identifyClient == null) + return; + + var task = _identifyClient.PeopleSetAsync(analyticsId, traits); + task.ContinueWith(t => + { + if (t.Exception != null) + Debug.WriteLine("MixpanelClient.Identify: best-effort identify failed: " + + t.Exception); + }, TaskContinuationOptions.OnlyOnFaulted); + } + catch (Exception e) + { + Debug.WriteLine("MixpanelClient.Identify failed: " + e); + } } + /// + /// Scrubs the event name and every string property value (this is where stack traces -- + /// spooled as the "Stack Trace" property of the "Exception" event -- get their embedded user + /// paths normalized), stamps $insert_id/time via , and + /// enqueues into the spool. An event over Mixpanel's per-event size limit (see + /// ) is refused at enqueue and counted in + /// . Never throws once initialized; calling it on a client + /// that was never initialized at all is a programming error in the host and throws + /// (matching the original MixpanelClient, which dereferenced a null field in that case). + /// public void Track(string analyticsId, string eventName, JsonObject properties) { - _tasks.Add(_client.TrackAsync(eventName, analyticsId, properties)); + // Deliberately OUTSIDE the catch-all below: a never-initialized client must surface the + // programming error rather than silently dropping every event. Distinct from _spool == + // null, which means Initialize ran but the spool could not be created (e.g. another + // instance holds the cross-process lock) -- that case degrades to a no-op by design. + if (!_initialized) + throw new InvalidOperationException( + "MixpanelClient.Track called before Initialize"); + + try + { + if (_spool == null) + return; + + var scrubbedName = PathScrubber.Scrub(eventName); + var scrubbedProperties = ScrubProperties(properties); + + var evt = AnalyticsEvent.Create( + analyticsId, + scrubbedName, + scrubbedProperties, + time: _timeProvider.GetUtcNow()); + + Interlocked.Increment(ref _submitted); + if (!_spool.Enqueue(evt)) + { + // Dropped at enqueue (most likely over the per-event size cap; see + // kMaxSpooledEventBytes) -- surface it in the statistics like any other + // undeliverable event. + Interlocked.Increment(ref _failed); + } + } + catch (Exception e) + { + Debug.WriteLine("MixpanelClient.Track failed: " + e); + } } - // Flush is really a no-op in our current Mixpanel client + private static JsonObject ScrubProperties(JsonObject properties) + { + var result = new JsonObject(); + if (properties == null) + return result; + + foreach (var kv in properties) + { + if (kv.Value is JsonPrimitive primitive && primitive.IsString) + result[kv.Key] = PathScrubber.Scrub(primitive.Content); + else + result[kv.Key] = kv.Value; + } + + return result; + } + + /// + /// Bounded drain: repeatedly attempts a drain until the spool is empty or a bound + /// (wall-clock duration or attempt count) is hit, then returns. Used by both + /// and so an offline flush/shutdown returns fast + /// instead of hanging the host. + /// + /// + /// Two things make this provably bounded even against a "black hole" server (one that + /// accepts the TCP connection but never responds -- captive portals and some firewalls do + /// this), which the plain path is not: + /// + /// A single covering the whole wall-clock + /// budget is shared by every attempt, passed all the way down to + /// via . Without + /// this, the deadline below is only checked BETWEEN attempts -- one hanging attempt could + /// run as long as the sender's own retry-multiplied HTTP timeout (e.g. 3 attempts x a 15s + /// HttpClient timeout ~= 45s) before the bound is ever checked. Created via the injected + /// (not new CancellationTokenSource(delay), which only + /// knows the system clock) so tests keep deterministic control of the deadline. + /// Retries are bypassed (usePipeline: false): retrying during a bounded + /// shutdown/flush has little delivery value (we are about to give up on this attempt + /// anyway) and would multiply -- rather than bound -- the time spent waiting on a hanging + /// server. + /// + /// + private async Task BoundedDrainAsync() + { + if (_spool == null) + return; + + try + { + using (var cts = _timeProvider.CreateCancellationTokenSource(s_boundedDrainDuration)) + { + for (var i = 0; i < kBoundedDrainMaxAttempts; i++) + { + if (cts.IsCancellationRequested) + return; + + if (_spool.ApproximateCount <= 0) + return; + + await DrainOnceCoreAsync(cts.Token, usePipeline: false).ConfigureAwait(false); + } + } + } + catch (Exception e) + { + Debug.WriteLine("MixpanelClient.BoundedDrain failed: " + e); + } + } + + /// Bounded drain; returns fast even while offline. See . + /// Never faults. + public async Task FlushAsync() + { + try + { + await BoundedDrainAsync().ConfigureAwait(false); + } + catch (Exception e) + { + Debug.WriteLine("MixpanelClient.Flush failed: " + e); + } + } + + /// Synchronous . + /// + /// Blocks on the bounded drain -- safe even from a UI thread because the entire drain path + /// awaits with ConfigureAwait(false) (no continuation ever needs the caller's + /// SynchronizationContext), and the drain itself is bounded to a few seconds. + /// public void Flush() { + try + { + FlushAsync().GetAwaiter().GetResult(); + } + catch (Exception e) + { + Debug.WriteLine("MixpanelClient.Flush failed: " + e); + } + } + + /// + /// Bounded drain, then release the spool's cross-process lock. Never hangs, even fully + /// offline: undelivered events simply remain on disk for the next launch to pick up. + /// Never faults, and is idempotent -- a redundant (e.g. Dispose on + /// the facade after an explicit ShutDownAsync) finds a stopped + /// timer, an empty-reporting spool, and a no-op re-Dispose. + /// + public async Task ShutDownAsync() + { + try + { + StopTimerPermanently(); + await BoundedDrainAsync().ConfigureAwait(false); + } + catch (Exception e) + { + Debug.WriteLine("MixpanelClient.ShutDown failed: " + e); + } + finally + { + try + { + _spool?.Dispose(); + } + catch (Exception e) + { + Debug.WriteLine("MixpanelClient.ShutDown: failed to dispose spool: " + e); + } + } + } + + /// Synchronous . See for why + /// blocking here is safe. + public void ShutDown() + { + try + { + ShutDownAsync().GetAwaiter().GetResult(); + } + catch (Exception e) + { + Debug.WriteLine("MixpanelClient.ShutDown failed: " + e); + } + } + + /// + /// Consent revocation: empties the spool immediately and pauses the flush loop (the timer is + /// stopped and opportunistic network-change kicks are ignored). This is a pause, not a teardown: + /// if consent is granted again within the same process run, Analytics.AllowTracking's + /// setter calls to re-arm the loop. + /// + public void PurgeQueuedEvents() + { + try + { + PauseTimer(); + _spool?.Purge(); + } + catch (Exception e) + { + Debug.WriteLine("MixpanelClient.PurgeQueuedEvents failed: " + e); + } } - public Statistics Statistics => new Statistics(_tasks.Count, - _tasks.Count(t => t.IsCompleted && t.Result), _tasks.Count(t => t.IsCompleted && !t.Result)); + /// + /// Re-arms the flush loop after paused it (consent revoked then + /// granted again in the same run). Clears the paused flag and reschedules the timer to fire + /// shortly so newly tracked events go out promptly. No-op on the timer if the client was never + /// initialized (e.g. deferred-init clients) or has been shut down. + /// + public void ResumeSending() + { + try + { + // Clear the flag unconditionally so the paused state is correct even for a client whose + // timer was never started (deferred init); reschedule only if there is a live timer. + _paused = false; + _flushTimer?.Change( + TimeSpan.FromSeconds(kReconnectFlushDelaySeconds), _flushInterval); + } + catch (Exception e) + { + Debug.WriteLine("MixpanelClient.ResumeSending failed: " + e); + } + } + + private void PauseTimer() + { + try + { + _paused = true; + _flushTimer?.Change(Timeout.Infinite, Timeout.Infinite); + } + catch (Exception e) + { + Debug.WriteLine("MixpanelClient: failed to pause timer: " + e); + } + } + + private void StopTimerPermanently() + { + try + { + _paused = true; + UnsubscribeFromNetworkChanges(); + _flushTimer?.Change(Timeout.Infinite, Timeout.Infinite); + _flushTimer?.Dispose(); + _flushTimer = null; + } + catch (Exception e) + { + Debug.WriteLine("MixpanelClient: failed to stop timer: " + e); + } + } + + public Statistics Statistics => new Statistics(_submitted, _succeeded, _failed); + + // Test seam: exposes whether the flush loop is currently paused (consent revoked). See + // MixpanelClientTests. Not part of IClient. + internal bool SendingPaused => _paused; } -} \ No newline at end of file +} diff --git a/src/DesktopAnalytics/MixpanelEventSender.cs b/src/DesktopAnalytics/MixpanelEventSender.cs new file mode 100644 index 0000000..02a4f60 --- /dev/null +++ b/src/DesktopAnalytics/MixpanelEventSender.cs @@ -0,0 +1,220 @@ +using System; +using System.Diagnostics; +using System.Net.Http; +using System.Text; +using System.Threading; +using System.Threading.Tasks; +using Segment.Serialization; + +namespace DesktopAnalytics +{ + /// + /// The real (network-hitting) : posts one event at a time to Mixpanel's + /// HTTP "track" endpoint (https://developer.mixpanel.com/reference/track-event) and classifies the + /// response into a for the flush loop. + /// + /// + /// Deliberately does not go through the mixpanel-csharp package for this path: that package's + /// TrackAsync reports success/failure as a single bool, which is not enough information + /// to distinguish "retry me" from "poison, drop me" (see offline-analytics.md, "Failure + /// classification"). Posting the well-documented raw HTTP form directly keeps that distinction and + /// makes the base URL trivially swappable for tests (WireMock.Net). mixpanel-csharp is still + /// used, elsewhere, for the best-effort (non-spooled) Identify call. + /// + internal class MixpanelEventSender : IEventSender, IDisposable + { + internal const string kDefaultBaseUrl = "https://api.mixpanel.com"; + + private readonly string _apiSecret; + private readonly string _baseUrl; + private readonly HttpClient _httpClient; + private readonly bool _ownsHttpClient; + + /// The Mixpanel project token. + /// Base URL of the Mixpanel HTTP API. Defaults to the real Mixpanel + /// endpoint; tests override this with a WireMock.Net base URL. + /// Injectable so tests can supply an already + /// configured (e.g. short timeout) or point it at a local stub. If omitted, this instance owns + /// and disposes its own . + public MixpanelEventSender(string apiSecret, string baseUrl = kDefaultBaseUrl, HttpClient httpClient = null) + { + _apiSecret = apiSecret; + _baseUrl = string.IsNullOrEmpty(baseUrl) + ? kDefaultBaseUrl + : baseUrl.TrimEnd('/'); + + if (httpClient != null) + { + _httpClient = httpClient; + _ownsHttpClient = false; + } + else + { + // AllowAutoRedirect is disabled deliberately: with the default handler (which follows + // redirects), a captive portal (hotel/coffee-shop Wi-Fi login page) that intercepts + // this POST would 302 us to its login page, HttpClient would follow it, get back a + // harmless 200, and the code below would classify that as Delivered -- silently + // losing the event even though Mixpanel never received it. With redirects disabled we + // see the raw 3xx instead and can classify it correctly (see the status-code check + // below). + _httpClient = new HttpClient(new HttpClientHandler { AllowAutoRedirect = false }) + { Timeout = TimeSpan.FromSeconds(15) }; + _ownsHttpClient = true; + } + } + + /// + /// Posts to Mixpanel's /track endpoint and classifies the result. Never + /// throws (and never returns a faulted task): any serialization, network, or cancellation + /// failure is caught and reported as a (see class remarks for the + /// classification rules). + /// + /// See . A bounded caller + /// ('s BoundedDrain) uses this to abandon a request against a + /// server that accepts the connection but never responds; that is reported as + /// , exactly like any other transient failure. + public async Task SendAsync(AnalyticsEvent evt, CancellationToken cancellationToken = default) + { + try + { + if (evt == null) + return SendResult.PoisonDrop; + + string json; + try + { + json = BuildPayloadJson(evt); + } + catch (Exception e) + { + // A payload we cannot even serialize will never succeed -- drop it rather than + // retry forever. + Debug.WriteLine("MixpanelEventSender.Send: failed to build payload, dropping: " + e); + return SendResult.PoisonDrop; + } + + var formBody = "data=" + Uri.EscapeDataString(json); + + HttpResponseMessage response; + try + { + using (var content = new StringContent(formBody, Encoding.UTF8, + "application/x-www-form-urlencoded")) + { + response = await _httpClient + .PostAsync(_baseUrl + "/track", content, cancellationToken) + .ConfigureAwait(false); + } + } + catch (OperationCanceledException e) + { + // The caller's deadline (see MixpanelClient.BoundedDrain) fired before the server + // responded -- most commonly a "black hole" server that accepts the TCP connection + // but never replies (captive portals and some firewalls do this). Treat exactly + // like any other transient failure: leave the event in the spool for a later + // retry rather than throwing into (and hanging) the caller. + Debug.WriteLine("MixpanelEventSender.Send: canceled (deadline), will retry: " + e); + return SendResult.RetryableFailure; + } + catch (Exception e) + { + // Connection refused, DNS failure, timeout, etc. -- transient from our point of + // view, so leave the event in the spool for a later retry. + Debug.WriteLine("MixpanelEventSender.Send: network failure, will retry: " + e); + return SendResult.RetryableFailure; + } + + using (response) + { + var status = (int)response.StatusCode; + if (status >= 200 && status < 300) + { + // Mixpanel's /track returns HTTP 200 with a plain-text body of "0" when it + // rejects the payload (e.g. malformed data) -- as opposed to "1" for a genuine + // accept. A "0" will never succeed on retry, so treat it as poison rather than + // Delivered so it cannot silently vanish while never actually landing. + try + { + var body = await response.Content.ReadAsStringAsync() + .ConfigureAwait(false); + if (body != null && body.Trim() == "0") + return SendResult.PoisonDrop; + } + catch (Exception e) + { + // A body-read failure on an otherwise-successful response shouldn't turn a + // delivered event into a retry loop -- fall through and count it Delivered. + // At-least-once semantics + $insert_id dedup make an occasional + // false-positive here harmless. + Debug.WriteLine( + "MixpanelEventSender.Send: failed to read 2xx response body, assuming delivered: " + + e); + } + + return SendResult.Delivered; + } + + if (status >= 300 && status < 400) + { + // With AllowAutoRedirect disabled (see the constructor), a 3xx here means + // something intercepted the POST before it reached Mixpanel -- most commonly a + // captive portal redirecting to its login page. Mixpanel's /track endpoint + // never legitimately redirects, so treat this as an intercepting middlebox: + // leave the event spooled for retry instead of risking a misclassification. + return SendResult.RetryableFailure; + } + + if (status == 408 || status == 429 || status >= 500) + return SendResult.RetryableFailure; + + // Any other 4xx (or any other unexpected status) will never succeed by retrying -- + // treat it as a poison message so it cannot wedge the spool. + return SendResult.PoisonDrop; + } + } + catch (OperationCanceledException e) + { + Debug.WriteLine("MixpanelEventSender.Send: canceled (deadline), will retry: " + e); + return SendResult.RetryableFailure; + } + catch (Exception e) + { + // Belt-and-suspenders: analytics must never throw into the host. Treat anything + // unexpected as retryable rather than silently dropping the event. + Debug.WriteLine("MixpanelEventSender.Send failed unexpectedly, will retry: " + e); + return SendResult.RetryableFailure; + } + } + + private string BuildPayloadJson(AnalyticsEvent evt) + { + var properties = new JsonObject + { + { "token", _apiSecret }, + { "distinct_id", evt.AnalyticsId }, + { "$insert_id", evt.InsertId }, + { "time", evt.Time.ToUnixTimeSeconds() } + }; + + if (evt.Properties != null) + { + foreach (var kv in evt.Properties) + properties[kv.Key] = kv.Value; + } + + var payload = new JsonObject + { + { "event", evt.EventName }, + { "properties", properties } + }; + + return JsonUtility.ToJson(payload, false); + } + + public void Dispose() + { + if (_ownsHttpClient) + _httpClient?.Dispose(); + } + } +} diff --git a/src/DesktopAnalytics/PathScrubber.cs b/src/DesktopAnalytics/PathScrubber.cs new file mode 100644 index 0000000..c4475e6 --- /dev/null +++ b/src/DesktopAnalytics/PathScrubber.cs @@ -0,0 +1,39 @@ +using System; +using System.Text.RegularExpressions; + +namespace DesktopAnalytics +{ + /// + /// Normalizes user home paths embedded in strings (e.g. exception stack traces) so that the + /// OS account name does not leak when spooled/reported. This is not a general PII scrubber; + /// it only targets the well-known Windows and Unix "home directory" prefixes. + /// + internal static class PathScrubber + { + // Windows: C:\Users\\ (any drive letter, case-insensitive on both the drive + // letter and the literal "Users" segment). We stop matching the user name at the next + // path separator so we don't eat the rest of the path. + private static readonly Regex s_windowsUserPath = + new Regex(@"[A-Za-z]:\\Users\\[^\\/:*?""<>|\r\n]+\\", + RegexOptions.Compiled | RegexOptions.IgnoreCase); + + // Unix: /home// + private static readonly Regex s_unixUserPath = + new Regex(@"/home/[^/\r\n]+/", RegexOptions.Compiled); + + /// + /// Replaces every occurrence of a Windows or Unix user-home-directory prefix in + /// with a neutral placeholder ("%USER%\" or "%USER%/" + /// respectively). Returns the input unchanged if it is null/empty or contains no match. + /// + public static string Scrub(string input) + { + if (string.IsNullOrEmpty(input)) + return input; + + var result = s_windowsUserPath.Replace(input, @"%USER%\"); + result = s_unixUserPath.Replace(result, "%USER%/"); + return result; + } + } +} diff --git a/src/DesktopAnalytics/SegmentClient.cs b/src/DesktopAnalytics/SegmentClient.cs index 0a53f18..d0b58a7 100644 --- a/src/DesktopAnalytics/SegmentClient.cs +++ b/src/DesktopAnalytics/SegmentClient.cs @@ -1,5 +1,6 @@ using System; using System.Diagnostics; +using System.Threading.Tasks; using Segment.Concurrent; using Segment.Serialization; @@ -67,9 +68,44 @@ public void Flush() _analytics.Flush(); } + /// + /// Completes synchronously: Segment.Analytics.CSharp's Flush() only signals the + /// library's own background delivery (its coroutine system) and exposes nothing awaitable, + /// so there is no async work to represent here. + /// + public Task ShutDownAsync() + { + ShutDown(); + return Task.CompletedTask; + } + + /// See for why this completes synchronously. + public Task FlushAsync() + { + Flush(); + return Task.CompletedTask; + } + public Statistics Statistics => new Statistics(StatMonitor.Submitted, StatMonitor.Succeeded, StatMonitor.Failed); + /// + /// The Segment path already gets offline durability from Segment.Analytics.CSharp's own + /// on-disk storage/retry, and this class does not maintain a separate spool of its own -- so + /// there is nothing here to purge on consent revocation. + /// + public void PurgeQueuedEvents() + { + } + + /// + /// No-op: this client has no separate background flush loop of its own to re-arm (the + /// Segment.Analytics.CSharp library manages its own delivery), so there is nothing to resume. + /// + public void ResumeSending() + { + } + public void OnExceptionThrown(Exception e) { Debug.WriteLine($"**** Segment.IO Failed to deliver. {e.Message}"); diff --git a/src/DesktopAnalyticsTests/AnalyticsEventTests.cs b/src/DesktopAnalyticsTests/AnalyticsEventTests.cs new file mode 100644 index 0000000..5065b47 --- /dev/null +++ b/src/DesktopAnalyticsTests/AnalyticsEventTests.cs @@ -0,0 +1,132 @@ +using System; +using DesktopAnalytics; +using NUnit.Framework; +using Segment.Serialization; + +namespace DesktopAnalyticsTests +{ + [TestFixture] + public class AnalyticsEventTests + { + private static string ContentOf(JsonElement element) + { + return element.ToJsonPrimitive().Content; + } + + [Test] + public void ToBytes_FromBytes_RoundTrip_PreservesAllScalarFields() + { + var time = new DateTimeOffset(2026, 7, 8, 12, 34, 56, TimeSpan.Zero); + var original = new AnalyticsEvent + { + AnalyticsId = "abc-123", + EventName = "Save PDF", + Properties = new JsonObject { { "Portion", "All" } }, + InsertId = "insert-guid-1", + Time = time + }; + + var roundTripped = AnalyticsEvent.FromBytes(original.ToBytes()); + + Assert.AreEqual(original.AnalyticsId, roundTripped.AnalyticsId); + Assert.AreEqual(original.EventName, roundTripped.EventName); + Assert.AreEqual(original.InsertId, roundTripped.InsertId); + Assert.AreEqual(original.Time, roundTripped.Time); + } + + [Test] + public void ToBytes_ReturnsUtf8EncodedJson() + { + var evt = new AnalyticsEvent + { + AnalyticsId = "abc-123", + EventName = "Save PDF", + InsertId = "insert-guid-1", + Time = DateTimeOffset.UtcNow + }; + + var bytes = evt.ToBytes(); + var json = System.Text.Encoding.UTF8.GetString(bytes); + + StringAssert.Contains("\"abc-123\"", json); + StringAssert.Contains("\"Save PDF\"", json); + } + + [Test] + public void PropertiesBag_WithMultipleKeys_SurvivesRoundTrip() + { + var original = new AnalyticsEvent + { + AnalyticsId = "abc-123", + EventName = "Exception", + Properties = new JsonObject + { + { "Message", "Oops" }, + { "Stack Trace", "at Foo.Bar()" }, + { "Count", 3 } + }, + InsertId = "insert-guid-2", + Time = DateTimeOffset.UtcNow + }; + + var roundTripped = AnalyticsEvent.FromBytes(original.ToBytes()); + + Assert.AreEqual(3, roundTripped.Properties.Count); + Assert.AreEqual("Oops", ContentOf(roundTripped.Properties["Message"])); + Assert.AreEqual("at Foo.Bar()", ContentOf(roundTripped.Properties["Stack Trace"])); + Assert.AreEqual("3", ContentOf(roundTripped.Properties["Count"])); + } + + [Test] + public void Create_WithNoInjectedDependencies_StampsRealGuidAndRecentTime() + { + var before = DateTimeOffset.UtcNow; + var evt = AnalyticsEvent.Create("abc-123", "Launch"); + var after = DateTimeOffset.UtcNow; + + Assert.IsTrue(Guid.TryParse(evt.InsertId, out _)); + Assert.GreaterOrEqual(evt.Time, before); + Assert.LessOrEqual(evt.Time, after); + } + + [Test] + public void Create_WithInjectedDependencies_StampsExactInjectedValues() + { + var fixedTime = new DateTimeOffset(2026, 1, 2, 3, 4, 5, TimeSpan.Zero); + const string fixedInsertId = "fixed-insert-id"; + + var evt = AnalyticsEvent.Create( + "abc-123", + "Launch", + new JsonObject { { "Version", "1.2.3" } }, + insertId: fixedInsertId, + time: fixedTime + ); + + Assert.AreEqual("abc-123", evt.AnalyticsId); + Assert.AreEqual("Launch", evt.EventName); + Assert.AreEqual(fixedInsertId, evt.InsertId); + Assert.AreEqual(fixedTime, evt.Time); + Assert.AreEqual("1.2.3", ContentOf(evt.Properties["Version"])); + } + + [Test] + public void Create_CalledTwice_ProducesDifferentInsertIdsWhenNotInjected() + { + var first = AnalyticsEvent.Create("abc-123", "Launch"); + var second = AnalyticsEvent.Create("abc-123", "Launch"); + + Assert.AreNotEqual(first.InsertId, second.InsertId); + } + + [Test] + public void Create_WithNullProperties_ProducesEmptyPropertyBag() + { + var evt = AnalyticsEvent.Create("abc-123", "Launch", null, + "insert-id", DateTimeOffset.UtcNow); + + Assert.IsNotNull(evt.Properties); + Assert.AreEqual(0, evt.Properties.Count); + } + } +} diff --git a/src/DesktopAnalyticsTests/DesktopAnalyticsTests.csproj b/src/DesktopAnalyticsTests/DesktopAnalyticsTests.csproj index 42efff0..e35283b 100644 --- a/src/DesktopAnalyticsTests/DesktopAnalyticsTests.csproj +++ b/src/DesktopAnalyticsTests/DesktopAnalyticsTests.csproj @@ -6,12 +6,17 @@ SIL International False + + true + + diff --git a/src/DesktopAnalyticsTests/EventSpoolTests.cs b/src/DesktopAnalyticsTests/EventSpoolTests.cs new file mode 100644 index 0000000..4493dc4 --- /dev/null +++ b/src/DesktopAnalyticsTests/EventSpoolTests.cs @@ -0,0 +1,578 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Threading; +using System.Threading.Tasks; +using DesktopAnalytics; +using NUnit.Framework; + +namespace DesktopAnalyticsTests +{ + // NOTE on the "corrupt/partial spool" edge case (power loss mid-write): rather than + // hand-crafting a truncated transaction-log file matching DiskQueue's private on-disk format + // (which would couple this suite to internals that may change between DiskQueue versions), + // test 10 below appends format-agnostic garbage bytes to the transaction log and asserts the + // spool survives -- exercising DiskQueue's transactional recovery (entries are only + // discoverable after a matching start/end transaction marker pair) without depending on what + // the markers look like. Separately, EventSpool.ProcessBatch guards against an individual + // entry that deserializes badly (e.g. an unreadable/incompatible schema version) by flushing + // (dropping) it like a poison message rather than wedging the spool -- see EventSpool.cs. + [TestFixture] + public class EventSpoolTests + { + private string _spoolDir; + + [SetUp] + public void SetUp() + { + _spoolDir = Path.Combine(Path.GetTempPath(), "EventSpoolTests_" + Guid.NewGuid()); + } + + [TearDown] + public void TearDown() + { + // Any EventSpool opened by a test must already be disposed (releasing DiskQueue's + // exclusive lock) by the time we get here -- each test disposes its spool(s) via + // `using` before returning. + if (Directory.Exists(_spoolDir)) + { + try + { + Directory.Delete(_spoolDir, true); + } + catch (Exception e) + { + // Best-effort cleanup; don't fail the test run over a leftover temp directory. + Console.WriteLine("EventSpoolTests.TearDown: failed to delete " + _spoolDir + ": " + e); + } + } + } + + private static AnalyticsEvent MakeEvent(string name) + { + return AnalyticsEvent.Create("user-1", name); + } + + // Reads every file under the spool directory that can be opened for shared reading. + // DiskQueue holds its 'lock' file open exclusively while the spool is open; that file + // only ever contains a process id, never event data, so skipping unopenable files does + // not weaken any event-content assertion. + internal static IEnumerable> ReadableSpoolFiles(string spoolDir) + { + foreach (var file in Directory.GetFiles(spoolDir, "*", SearchOption.AllDirectories)) + { + string content; + try + { + using (var stream = new FileStream(file, FileMode.Open, FileAccess.Read, + FileShare.ReadWrite | FileShare.Delete)) + using (var reader = new StreamReader(stream, System.Text.Encoding.UTF8)) + content = reader.ReadToEnd(); + } + catch (IOException) + { + continue; // DiskQueue's exclusively-held lock file. + } + + yield return new KeyValuePair(file, content); + } + } + + // 1. RESTART DURABILITY: enqueue N, dispose, reopen a NEW spool on the same directory -- + // all N survive, and a subsequent ProcessBatch (all Delivered) empties it. + [Test] + public async Task Enqueue_ThenDisposeAndReopen_EventsSurviveAndDrainFully() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + for (var i = 0; i < 5; i++) + spool.Enqueue(MakeEvent("Event-" + i)); + } // Disposed here, releasing the exclusive lock. + + using (var reopened = new EventSpool(_spoolDir, 10)) + { + Assert.AreEqual(5, reopened.ApproximateCount); + + var delivered = new List(); + await reopened.ProcessBatchAsync(10, evt => + { + delivered.Add(evt.EventName); + return Task.FromResult(SendResult.Delivered); + }); + + Assert.AreEqual(5, delivered.Count); + Assert.AreEqual(0, reopened.ApproximateCount); + } + } + + // 2. ProcessBatch where send returns Delivered for all => queue empty afterward. + [Test] + public async Task ProcessBatch_AllDelivered_EmptiesSpool() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + spool.Enqueue(MakeEvent("A")); + spool.Enqueue(MakeEvent("B")); + + await spool.ProcessBatchAsync(10, evt => Task.FromResult(SendResult.Delivered)); + + Assert.AreEqual(0, spool.ApproximateCount); + } + } + + // 3. send returns RetryableFailure on the 2nd item => 1st removed, 2nd and rest remain; + // a later ProcessBatch delivers them. + [Test] + public async Task ProcessBatch_RetryableFailureOnSecondItem_LeavesSecondAndRestInSpool() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + spool.Enqueue(MakeEvent("A")); + spool.Enqueue(MakeEvent("B")); + spool.Enqueue(MakeEvent("C")); + + var firstPassCalls = new List(); + await spool.ProcessBatchAsync(10, evt => + { + firstPassCalls.Add(evt.EventName); + return Task.FromResult(evt.EventName == "B" ? SendResult.RetryableFailure : SendResult.Delivered); + }); + + // A was delivered and flushed; B failed (not flushed, so processing stopped before C. + CollectionAssert.AreEqual(new[] { "A", "B" }, firstPassCalls); + Assert.AreEqual(2, spool.ApproximateCount); + + var secondPassCalls = new List(); + await spool.ProcessBatchAsync(10, evt => + { + secondPassCalls.Add(evt.EventName); + return Task.FromResult(SendResult.Delivered); + }); + + CollectionAssert.AreEqual(new[] { "B", "C" }, secondPassCalls); + Assert.AreEqual(0, spool.ApproximateCount); + } + } + + // 4. CRASH-WINDOW: send records the event then THROWS => nothing removed; reopening the + // spool shows the same events still present (at-least-once, no data loss). + [Test] + public async Task ProcessBatch_SendThrows_NothingRemovedAndEventsSurviveReopen() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + spool.Enqueue(MakeEvent("Crash-1")); + spool.Enqueue(MakeEvent("Crash-2")); + + var recorded = new List(); + Assert.DoesNotThrowAsync(async () => await spool.ProcessBatchAsync(10, evt => + { + // Simulate a sender that actually delivered the event over the network, then + // crashed/threw before this method could observe success and flush. + recorded.Add(evt.EventName); + throw new InvalidOperationException("simulated crash after send, before ack"); + })); + + Assert.AreEqual(new[] { "Crash-1" }, recorded); + Assert.AreEqual(2, spool.ApproximateCount); + } + + using (var reopened = new EventSpool(_spoolDir, 10)) + { + Assert.AreEqual(2, reopened.ApproximateCount); + + var delivered = new List(); + await reopened.ProcessBatchAsync(10, evt => + { + delivered.Add(evt.EventName); + return Task.FromResult(SendResult.Delivered); + }); + + CollectionAssert.AreEqual(new[] { "Crash-1", "Crash-2" }, delivered); + } + } + + // 5. Poison: send returns PoisonDrop => item removed even though not delivered. + [Test] + public async Task ProcessBatch_PoisonDrop_RemovesItemEvenThoughNotDelivered() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + spool.Enqueue(MakeEvent("Poison")); + spool.Enqueue(MakeEvent("Good")); + + var delivered = new List(); + await spool.ProcessBatchAsync(10, evt => + { + if (evt.EventName == "Poison") + return Task.FromResult(SendResult.PoisonDrop); + delivered.Add(evt.EventName); + return Task.FromResult(SendResult.Delivered); + }); + + CollectionAssert.AreEqual(new[] { "Good" }, delivered); + Assert.AreEqual(0, spool.ApproximateCount); + } + } + + // 6. Bounding: maxItems=3, enqueue 5 => count==3, the 3 NEWEST remain. + [Test] + public async Task Enqueue_ExceedingMaxItems_DropsOldestKeepsNewest() + { + using (var spool = new EventSpool(_spoolDir, 3)) + { + for (var i = 0; i < 5; i++) + spool.Enqueue(MakeEvent("Event-" + i)); + + Assert.AreEqual(3, spool.ApproximateCount); + + var remaining = new List(); + await spool.ProcessBatchAsync(10, evt => + { + remaining.Add(evt.EventName); + return Task.FromResult(SendResult.Delivered); + }); + + CollectionAssert.AreEqual(new[] { "Event-2", "Event-3", "Event-4" }, remaining); + } + } + + // 7. Purge empties a non-empty spool. + [Test] + public async Task Purge_EmptiesNonEmptySpool() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + for (var i = 0; i < 5; i++) + spool.Enqueue(MakeEvent("Event-" + i)); + + Assert.AreEqual(5, spool.ApproximateCount); + + spool.Purge(); + + Assert.AreEqual(0, spool.ApproximateCount); + + // A drain after Purge should find nothing to send. + var delivered = new List(); + await spool.ProcessBatchAsync(10, evt => + { + delivered.Add(evt.EventName); + return Task.FromResult(SendResult.Delivered); + }); + Assert.AreEqual(0, delivered.Count); + } + } + + // 7b. Purge leaves the spool fully usable: new events can be enqueued in the same + // session, only they survive a dispose/reopen, and the purged events' bytes are gone + // from the on-disk files once the spool is closed (the privacy property of a consent + // purge -- DiskQueue trims consumed entries rather than retaining them). + [Test] + public async Task Purge_ThenEnqueue_SpoolRemainsUsableAndPurgedBytesAreGoneFromDisk() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + spool.Enqueue(MakeEvent("PrePurgeSecret")); + Assert.AreEqual(1, spool.ApproximateCount); + + spool.Purge(); + Assert.AreEqual(0, spool.ApproximateCount); + + spool.Enqueue(MakeEvent("PostPurge")); + Assert.AreEqual(1, spool.ApproximateCount); + } + + foreach (var file in ReadableSpoolFiles(_spoolDir)) + { + StringAssert.DoesNotContain("PrePurgeSecret", file.Value, + "a consent purge must remove event data from disk, not just mark it consumed: " + file.Key); + } + + using (var reopened = new EventSpool(_spoolDir, 10)) + { + var delivered = new List(); + await reopened.ProcessBatchAsync(10, evt => + { + delivered.Add(evt.EventName); + return Task.FromResult(SendResult.Delivered); + }); + CollectionAssert.AreEqual(new[] { "PostPurge" }, delivered); + } + } + + // 8. Concurrent enqueue from multiple threads (single process): 4 threads x 25 events => + // all 100 persisted. Dedicated Threads (not Task.Run) so the intended parallelism is + // explicit rather than at the mercy of thread-pool scheduling. + [Test] + public async Task Enqueue_FromMultipleThreadsConcurrently_AllEventsPersisted() + { + const int threadCount = 4; + const int perThread = 25; + + using (var spool = new EventSpool(_spoolDir, threadCount * perThread)) + { + var threads = new Thread[threadCount]; + for (var t = 0; t < threadCount; t++) + { + var threadIndex = t; + threads[t] = new Thread(() => + { + for (var i = 0; i < perThread; i++) + spool.Enqueue(MakeEvent($"T{threadIndex}-{i}")); + }); + } + + foreach (var thread in threads) + thread.Start(); + foreach (var thread in threads) + thread.Join(); + + Assert.AreEqual(threadCount * perThread, spool.ApproximateCount); + + // Confirm every single one is actually retrievable, not just counted. + var seen = new HashSet(); + await spool.ProcessBatchAsync(threadCount * perThread, evt => + { + seen.Add(evt.EventName); + return Task.FromResult(SendResult.Delivered); + }); + Assert.AreEqual(threadCount * perThread, seen.Count); + } + } + + // 8a. Per-event byte cap: an event whose serialized form exceeds maxItemBytes is refused + // (Enqueue returns false) and never spooled, while a normal event on the same spool is + // accepted -- and Enqueue's return value distinguishes the two. + [Test] + public void Enqueue_EventLargerThanMaxItemBytes_IsRefusedWhileNormalEventIsAccepted() + { + using (var spool = new EventSpool(_spoolDir, 10, maxItemBytes: 500)) + { + var oversized = AnalyticsEvent.Create("user-1", "Huge", + new Segment.Serialization.JsonObject + { + { "Stack Trace", new string('x', 2000) } + }); + + Assert.IsFalse(spool.Enqueue(oversized), "an event over the byte cap must be refused"); + Assert.AreEqual(0, spool.ApproximateCount); + + Assert.IsTrue(spool.Enqueue(MakeEvent("Normal")), "a normal event must still be accepted"); + Assert.AreEqual(1, spool.ApproximateCount); + } + } + + // 8a2. Byte-based spool cap: enqueuing past maxSpoolBytes drops the oldest event(s), + // exactly like the item cap. + [Test] + public async Task Enqueue_ExceedingMaxSpoolBytes_DropsOldestKeepsNewest() + { + var events = new[] { MakeEvent("Event-A"), MakeEvent("Event-B"), MakeEvent("Event-C"), MakeEvent("Event-D") }; + // Cap sized (from the real serialized lengths) to hold the first three but not all four. + long cap = events[0].ToBytes().Length + events[1].ToBytes().Length + events[2].ToBytes().Length; + + using (var spool = new EventSpool(_spoolDir, 100, maxSpoolBytes: cap)) + { + foreach (var evt in events) + spool.Enqueue(evt); + + var remaining = new List(); + await spool.ProcessBatchAsync(10, evt => + { + remaining.Add(evt.EventName); + return Task.FromResult(SendResult.Delivered); + }); + + CollectionAssert.AreEqual(new[] { "Event-B", "Event-C", "Event-D" }, remaining, + "the byte cap must drop the OLDEST event to make room"); + } + } + + // 8a3. The byte accounting behind the cap survives a restart (measured from the live + // entries on open), so a reopened spool enforces the cap correctly. + [Test] + public void ApproximateBytes_SurvivesDisposeAndReopen() + { + long expectedBytes; + using (var spool = new EventSpool(_spoolDir, 10)) + { + spool.Enqueue(MakeEvent("Event-0")); + spool.Enqueue(MakeEvent("Event-1")); + expectedBytes = spool.ApproximateBytes; + Assert.Greater(expectedBytes, 0); + } + + using (var reopened = new EventSpool(_spoolDir, 10)) + { + Assert.AreEqual(expectedBytes, reopened.ApproximateBytes, + "the byte total must be re-measured from the live entries on open"); + Assert.AreEqual(2, reopened.ApproximateCount, "measuring must not consume the entries"); + } + } + + // 8a4. ProcessBatch byte budget: processing stops once the sent bytes reach maxBytes, but + // a single event over the whole budget still goes (forward progress is guaranteed). + [Test] + public async Task ProcessBatch_ByteBudget_StopsAfterBudgetButNeverStarves() + { + var events = new[] { MakeEvent("Event-0"), MakeEvent("Event-1"), MakeEvent("Event-2"), + MakeEvent("Event-3"), MakeEvent("Event-4") }; + // A budget the first two events fit under but the third pushes past. + long budget = events[0].ToBytes().Length + events[1].ToBytes().Length + 1; + + using (var spool = new EventSpool(_spoolDir, 10)) + { + foreach (var evt in events) + spool.Enqueue(evt); + + var firstCall = new List(); + await spool.ProcessBatchAsync(10, evt => + { + firstCall.Add(evt.EventName); + return Task.FromResult(SendResult.Delivered); + }, budget); + + CollectionAssert.AreEqual(new[] { "Event-0", "Event-1", "Event-2" }, firstCall, + "the budget is checked AFTER each event, so the event that crosses it is still sent"); + Assert.AreEqual(2, spool.ApproximateCount); + + // A budget smaller than any single event must still make progress, one event per call. + var secondCall = new List(); + await spool.ProcessBatchAsync(10, evt => + { + secondCall.Add(evt.EventName); + return Task.FromResult(SendResult.Delivered); + }, maxBytes: 1); + + CollectionAssert.AreEqual(new[] { "Event-3" }, secondCall); + Assert.AreEqual(1, spool.ApproximateCount); + } + } + + // 8b. ProcessBatch honors maxItems: with more events spooled than the batch size, exactly + // maxItems are attempted per call and the rest stay put for the next call. + [Test] + public async Task ProcessBatch_MoreEventsThanMaxItems_AttemptsExactlyMaxItemsPerCall() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + for (var i = 0; i < 5; i++) + spool.Enqueue(MakeEvent("Event-" + i)); + + var firstBatch = new List(); + await spool.ProcessBatchAsync(2, evt => + { + firstBatch.Add(evt.EventName); + return Task.FromResult(SendResult.Delivered); + }); + + CollectionAssert.AreEqual(new[] { "Event-0", "Event-1" }, firstBatch); + Assert.AreEqual(3, spool.ApproximateCount); + + var secondBatch = new List(); + await spool.ProcessBatchAsync(2, evt => + { + secondBatch.Add(evt.EventName); + return Task.FromResult(SendResult.Delivered); + }); + + CollectionAssert.AreEqual(new[] { "Event-2", "Event-3" }, secondBatch); + Assert.AreEqual(1, spool.ApproximateCount); + } + } + + // 9. Second EventSpool opened on the SAME directory while the first is still open + // fails/throws, proving the cross-process exclusive lock (DiskQueue.PersistentQueue.WaitFor + // with a short internal timeout) does not silently allow shared access. + [Test] + public void Constructor_SecondSpoolOnSameDirectoryWhileFirstOpen_Throws() + { + using (new EventSpool(_spoolDir, 10)) + { + // Assert.Catch (unlike Assert.Throws, which requires an *exact* type + // match) accepts any exception assignable to Exception -- we don't want this test + // coupled to DiskQueue's specific exception type (currently TimeoutException) for + // a lock-acquisition failure. + Assert.Catch(() => + { + using (new EventSpool(_spoolDir, 10)) + { + } + }); + } + } + + // 10. CORRUPTION, torn data-file write: garbage appended to a data file (bytes beyond the + // extents any committed transaction references) must not prevent reopening the spool, and + // events from committed transactions must still be readable. Appending garbage -- rather + // than hand-crafting file contents -- keeps this test independent of DiskQueue's private + // on-disk format (see the NOTE at the top of this fixture). + [Test] + public async Task Constructor_GarbageAppendedToDataFile_SpoolReopensAndCommittedEventsSurvive() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + spool.Enqueue(MakeEvent("Committed-1")); + spool.Enqueue(MakeEvent("Committed-2")); + } // Disposed: both enqueues are fully committed transactions on disk. + + AppendGarbageTo(FindSpoolFile("data.0")); + + using (var reopened = new EventSpool(_spoolDir, 10)) + { + var delivered = new List(); + await reopened.ProcessBatchAsync(10, evt => + { + delivered.Add(evt.EventName); + return Task.FromResult(SendResult.Delivered); + }); + CollectionAssert.AreEqual(new[] { "Committed-1", "Committed-2" }, delivered); + + // And the recovered spool must still accept new work. + reopened.Enqueue(MakeEvent("PostRecovery")); + Assert.AreEqual(1, reopened.ApproximateCount); + } + } + + // 10b. CORRUPTION, garbage in the transaction log: DiskQueue deliberately refuses to open + // a log with unrecognized trailing bytes (it recovers a TRUNCATED log -- the realistic + // power-loss shape, where entries simply end mid-transaction -- but treats garbage after a + // committed entry as a conflict). Pin that fail-fast behavior here: the constructor throws + // rather than silently serving corrupt data. At the client level this is already handled: + // MixpanelClient.Initialize catches the throw and degrades to a non-durable no-op client + // rather than crashing the host (see MixpanelClient's class remarks). + [Test] + public void Constructor_GarbageAppendedToTransactionLog_FailsFastRatherThanServingCorruptData() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + spool.Enqueue(MakeEvent("Committed-1")); + } + + AppendGarbageTo(FindSpoolFile("transaction.log")); + + Assert.Catch(() => + { + using (new EventSpool(_spoolDir, 10)) + { + } + }); + } + + private string FindSpoolFile(string fileName) + { + var path = Path.Combine(_spoolDir, fileName); + Assert.IsTrue(File.Exists(path), + "expected DiskQueue file at " + path + " -- if DiskQueue renamed it, update this test"); + return path; + } + + private static void AppendGarbageTo(string path) + { + var garbage = new byte[257]; + new Random(12345).NextBytes(garbage); + using (var stream = new FileStream(path, FileMode.Append, FileAccess.Write)) + stream.Write(garbage, 0, garbage.Length); + } + } +} diff --git a/src/DesktopAnalyticsTests/MixpanelClientTests.cs b/src/DesktopAnalyticsTests/MixpanelClientTests.cs new file mode 100644 index 0000000..ae0db3b --- /dev/null +++ b/src/DesktopAnalyticsTests/MixpanelClientTests.cs @@ -0,0 +1,771 @@ +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.IO; +using System.Threading; +using System.Threading.Tasks; +using DesktopAnalytics; +using NUnit.Framework; +using Segment.Serialization; + +namespace DesktopAnalyticsTests +{ + // Layer 3 of the fidelity ladder in offline-analytics.md: the real EventSpool (on a temp + // folder) driving a real MixpanelClient, with a scripted fake IEventSender standing in for the + // network. Tests await DrainOnceAsync() directly rather than racing the real background timer (see + // "Design-for-test seams" / "Flush-loop scheduling" in offline-analytics.md); InitializeForTest + // never starts that timer. + [TestFixture] + public class MixpanelClientTests + { + private string _spoolDir; + + [SetUp] + public void SetUp() + { + _spoolDir = Path.Combine(Path.GetTempPath(), "MixpanelClientTests_" + Guid.NewGuid()); + } + + [TearDown] + public void TearDown() + { + if (Directory.Exists(_spoolDir)) + { + try + { + Directory.Delete(_spoolDir, true); + } + catch (Exception e) + { + Console.WriteLine("MixpanelClientTests.TearDown: failed to delete " + _spoolDir + ": " + e); + } + } + } + + // A scripted IEventSender: each call to Send() consumes the next step in the script (or + // defaults to Delivered once the script is exhausted) and always records the event it was + // given, so tests can assert both what happened and what was actually handed to the sender. + private class ScriptedSender : IEventSender + { + private readonly Queue> _script; + public readonly List Sent = new List(); + + public ScriptedSender(IEnumerable> script) + { + _script = new Queue>(script); + } + + public Task SendAsync(AnalyticsEvent evt, CancellationToken cancellationToken = default) + { + Sent.Add(evt); + var step = _script.Count > 0 ? _script.Dequeue() : (_ => SendResult.Delivered); + // Deliberately NOT wrapped in Task.FromResult of a try/catch: a script step that + // throws makes this fake violate IEventSender's never-throw contract on purpose, + // exactly like the sync fake it replaced -- the crash-window tests rely on it. + return Task.FromResult(step(evt)); + } + } + + private class AlwaysResultSender : IEventSender + { + private readonly SendResult _result; + public int CallCount; + + public AlwaysResultSender(SendResult result) + { + _result = result; + } + + public Task SendAsync(AnalyticsEvent evt, CancellationToken cancellationToken = default) + { + CallCount++; + return Task.FromResult(_result); + } + } + + private class ThrowingSender : IEventSender + { + public Task SendAsync(AnalyticsEvent evt, CancellationToken cancellationToken = default) + { + throw new InvalidOperationException("simulated sender failure"); + } + } + + // Simulates a "black hole" server (accepts the connection but never responds): waits until + // its CancellationToken is signaled -- exactly what BoundedDrain's deadline does -- then + // reports the send as failed. Guards against ever being handed a token that can't be + // canceled, which would otherwise stall the whole test run rather than just fail one test. + private class BlockingUntilCanceledSender : IEventSender + { + public int CallCount; + + public async Task SendAsync(AnalyticsEvent evt, CancellationToken cancellationToken = default) + { + Interlocked.Increment(ref CallCount); + if (cancellationToken.CanBeCanceled) + { + try + { + await Task.Delay(TimeSpan.FromSeconds(10), cancellationToken); + } + catch (OperationCanceledException) + { + // The deadline fired -- fall through and report the failure, per the + // IEventSender contract (treat cancellation like any transient failure). + } + } + return SendResult.RetryableFailure; + } + } + + // ---- NO-LOSS ------------------------------------------------------------------------- + + [Test] + public async Task DrainOnce_RetryRetryThenDeliverAcrossSuccessiveCalls_DeliversExactlyOnceAndEmptiesSpool() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + var sender = new ScriptedSender(new Func[] + { + _ => SendResult.RetryableFailure, + _ => SendResult.RetryableFailure, + _ => SendResult.Delivered + }); + var client = new MixpanelClient(); + client.InitializeForTest(spool, sender); + + client.Track("user-1", "Save", null); + Assert.AreEqual(1, spool.ApproximateCount); + + await client.DrainOnceAsync(); // 1st attempt: RetryableFailure -- left in spool. + Assert.AreEqual(1, spool.ApproximateCount); + + await client.DrainOnceAsync(); // 2nd attempt: RetryableFailure -- still left in spool. + Assert.AreEqual(1, spool.ApproximateCount); + + await client.DrainOnceAsync(); // 3rd attempt: Delivered -- removed. + Assert.AreEqual(0, spool.ApproximateCount); + + Assert.AreEqual(3, sender.Sent.Count, "the event should have been handed to the sender exactly three times"); + Assert.AreEqual(1, client.Statistics.Succeeded); + Assert.AreEqual(0, client.Statistics.Failed); + } + } + + // ---- CRASH-WINDOW / DEDUP ------------------------------------------------------------- + + [Test] + public async Task DrainOnce_SenderRecordsThenThrows_SameInsertIdSeenAgainOnLaterRetry() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + var client = new MixpanelClient(); + + // A sender that behaves exactly like one that delivered the event over the network + // and then crashed/threw before this process could observe success and flush -- + // the crash-window this whole design exists to survive. + var crashingSender = new ScriptedSender(new Func[] + { + _ => throw new InvalidOperationException("simulated crash after send, before ack") + }); + client.InitializeForTest(spool, crashingSender); + client.Track("user-1", "Save", null); + + Assert.DoesNotThrowAsync(async () => await client.DrainOnceAsync()); + Assert.AreEqual(1, spool.ApproximateCount, "the event must not be removed -- the sender never confirmed delivery"); + Assert.AreEqual(1, crashingSender.Sent.Count); + + // "Reopen"/retry with a now-succeeding sender against the SAME spool. + var succeedingSender = new ScriptedSender(new Func[] + { + _ => SendResult.Delivered + }); + client.InitializeForTest(spool, succeedingSender); + await client.DrainOnceAsync(); + + Assert.AreEqual(0, spool.ApproximateCount); + Assert.AreEqual(1, succeedingSender.Sent.Count); + + // The key assertion: the sender saw the SAME $insert_id both times, proving Mixpanel + // would dedup the at-least-once replay rather than double-counting the event. + Assert.AreEqual(crashingSender.Sent[0].InsertId, succeedingSender.Sent[0].InsertId); + } + } + + // ---- POISON --------------------------------------------------------------------------- + + [Test] + public async Task DrainOnce_PoisonDrop_RemovesEventAndIncrementsFailedWithoutRetrying() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + var sender = new AlwaysResultSender(SendResult.PoisonDrop); + var client = new MixpanelClient(); + client.InitializeForTest(spool, sender); + + client.Track("user-1", "BadEvent", null); + await client.DrainOnceAsync(); + + Assert.AreEqual(0, spool.ApproximateCount, "a poison event must be dropped, not left to wedge the spool"); + Assert.AreEqual(1, sender.CallCount); + Assert.AreEqual(1, client.Statistics.Failed); + Assert.AreEqual(0, client.Statistics.Succeeded); + + // A further drain must not re-attempt it (it is already gone). + await client.DrainOnceAsync(); + Assert.AreEqual(1, sender.CallCount); + } + } + + // ---- CONSENT PURGE ---------------------------------------------------------------------- + + [Test] + public void PurgeQueuedEvents_EmptiesNonEmptySpool() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + var sender = new AlwaysResultSender(SendResult.Delivered); + var client = new MixpanelClient(); + client.InitializeForTest(spool, sender); + + client.Track("user-1", "A", null); + client.Track("user-1", "B", null); + Assert.AreEqual(2, spool.ApproximateCount); + + client.PurgeQueuedEvents(); + + Assert.AreEqual(0, spool.ApproximateCount); + } + } + + [Test] + public void PurgeQueuedEvents_NeverThrows_EvenWithNoSpool() + { + var client = new MixpanelClient(); + client.InitializeForTest(null, new AlwaysResultSender(SendResult.Delivered)); + + Assert.DoesNotThrow(() => client.PurgeQueuedEvents()); + } + + // ---- SCRUBBING -------------------------------------------------------------------------- + + [Test] + public async Task Track_ExceptionEventWithUserPathInStackTrace_ScrubsBeforeSpoolingAndSending() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + var sender = new AlwaysResultSender(SendResult.Delivered); + var client = new MixpanelClient(); + client.InitializeForTest(spool, sender); + + var props = new JsonObject + { + { "Message", "Boom" }, + { "Stack Trace", @"at Foo.Bar() in C:\Users\alice\src\Foo.cs:line 10" } + }; + client.Track("user-1", "Exception", props); + + // Prove it was scrubbed BEFORE hitting the disk by scanning the raw spool files + // themselves, before anything drains: no assertion on what the sender is later + // handed can distinguish scrub-at-enqueue from scrub-at-send, but the bytes on + // disk can. Guard against vacuously passing (e.g. if the persisted encoding ever + // changes) by requiring the scrubbed marker to actually be FOUND on disk. + var scrubbedMarkerFoundOnDisk = false; + foreach (var file in EventSpoolTests.ReadableSpoolFiles(_spoolDir)) + { + StringAssert.DoesNotContain("alice", file.Value, + "unscrubbed user path found on disk in " + file.Key); + if (file.Value.Contains("%USER%")) + scrubbedMarkerFoundOnDisk = true; + } + Assert.IsTrue(scrubbedMarkerFoundOnDisk, + "expected to find the scrubbed path (%USER%) in the raw spool files -- if this " + + "stops being readable as UTF-8 text, rework this scan"); + + // And the scrubbed form (not just an absent event) is what got persisted: drain + // with a sender that records what it was actually handed. + var scriptedSender = new ScriptedSender(new Func[] + { + _ => SendResult.Delivered + }); + var reopenedClient = new MixpanelClient(); + reopenedClient.InitializeForTest(spool, scriptedSender); + await reopenedClient.DrainOnceAsync(); + + Assert.AreEqual(1, scriptedSender.Sent.Count); + var sentStackTrace = ((JsonPrimitive)scriptedSender.Sent[0].Properties["Stack Trace"]).Content; + StringAssert.DoesNotContain("alice", sentStackTrace); + StringAssert.Contains("%USER%", sentStackTrace); + } + } + + [Test] + public async Task Track_EventNameContainingUserPath_ScrubsEventName() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + var client = new MixpanelClient(); + client.InitializeForTest(spool, new AlwaysResultSender(SendResult.Delivered)); + + client.Track("user-1", @"Opened C:\Users\alice\project.xyz", null); + + var sender = new ScriptedSender(new Func[] { _ => SendResult.Delivered }); + var drainer = new MixpanelClient(); + drainer.InitializeForTest(spool, sender); + await drainer.DrainOnceAsync(); + + StringAssert.DoesNotContain("alice", sender.Sent[0].EventName); + StringAssert.Contains("%USER%", sender.Sent[0].EventName); + } + } + + // ---- NEVER CRASH ------------------------------------------------------------------------ + + // Calling Track on a client that was never initialized at all is a programming error in + // the host and must throw (the original MixpanelClient dereferenced a null field there) -- + // NOT silently drop every event. Distinct from Track_WithNoSpool_NeverThrows below, where + // Initialize DID run but the spool could not be created. + [Test] + public void Track_WithoutInitialize_ThrowsInvalidOperationException() + { + var client = new MixpanelClient(); + Assert.Throws(() => client.Track("user-1", "Save", null)); + } + + [Test] + public void Track_WithNoSpool_NeverThrows() + { + var client = new MixpanelClient(); + // Simulates a failed Initialize (e.g. the EventSpool constructor threw acquiring the + // cross-process lock), which leaves the client with no spool at all. + client.InitializeForTest(null, new AlwaysResultSender(SendResult.Delivered)); + + Assert.DoesNotThrow(() => client.Track("user-1", "Save", null)); + Assert.DoesNotThrowAsync(async () => await client.DrainOnceAsync()); + Assert.DoesNotThrow(() => client.Flush()); + Assert.DoesNotThrowAsync(async () => await client.FlushAsync()); + Assert.DoesNotThrowAsync(async () => await client.ShutDownAsync()); + Assert.DoesNotThrow(() => client.ShutDown()); + } + + [Test] + public void DrainOnceAndFlush_SenderAlwaysThrows_NeverPropagateOutOfMixpanelClient() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + var client = new MixpanelClient(); + client.InitializeForTest(spool, new ThrowingSender()); + + client.Track("user-1", "Save", null); + + Assert.DoesNotThrowAsync(async () => await client.DrainOnceAsync()); + Assert.DoesNotThrow(() => client.Flush()); + + // A sender that only ever throws must not be treated as delivered/poison -- the + // event must still be sitting in the spool for a later retry. + Assert.AreEqual(1, spool.ApproximateCount); + } + } + + // ---- SHUTDOWN OFFLINE ------------------------------------------------------------------- + + [Test] + public void ShutDown_SenderAlwaysFails_ReturnsPromptlyAndEventsRemainOnDisk() + { + var spool = new EventSpool(_spoolDir, 10); + var sender = new AlwaysResultSender(SendResult.RetryableFailure); + var client = new MixpanelClient(); + client.InitializeForTest(spool, sender); + + for (var i = 0; i < 5; i++) + client.Track("user-1", "Save-" + i, null); + + var stopwatch = Stopwatch.StartNew(); + var completed = Task.Run(() => client.ShutDown()).Wait(TimeSpan.FromSeconds(15)); + stopwatch.Stop(); + + Assert.IsTrue(completed, "ShutDown() did not return within the timeout while offline"); + Assert.Less(stopwatch.Elapsed, TimeSpan.FromSeconds(15)); + + // ShutDown() must have released the spool's cross-process lock -- reopening it and + // finding all 5 events proves nothing was lost, and that the lock really was released. + using (var reopened = new EventSpool(_spoolDir, 10)) + { + Assert.AreEqual(5, reopened.ApproximateCount); + } + } + + [Test] + public void Flush_WhileOffline_ReturnsPromptlyWithoutEmptyingSpool() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + var sender = new AlwaysResultSender(SendResult.RetryableFailure); + var client = new MixpanelClient(); + client.InitializeForTest(spool, sender); + + client.Track("user-1", "Save", null); + + var stopwatch = Stopwatch.StartNew(); + var completed = Task.Run(() => client.Flush()).Wait(TimeSpan.FromSeconds(15)); + stopwatch.Stop(); + + Assert.IsTrue(completed, "Flush() did not return within the timeout while offline"); + Assert.AreEqual(1, spool.ApproximateCount, "an offline flush must leave the undelivered event on disk"); + } + } + + // ---- BOUNDED SHUTDOWN/FLUSH AGAINST A "BLACK HOLE" SERVER -------------------------------- + // + // Regression coverage for the bug where BoundedDrain's 5s wall-clock bound was only checked + // BETWEEN DrainOnce calls: a server that accepts the connection but never responds could + // make a single DrainOnce take up to ~45s (3 attempts x a 15s HttpClient timeout) before the + // bound was ever consulted, hanging ShutDown()/Flush(). These use a sender that blocks until + // the CancellationToken BoundedDrain hands it is canceled, which only happens promptly if + // the per-attempt deadline is actually wired through. + + [Test] + public void ShutDown_SenderBlocksUntilCanceled_ReturnsWithinBoundAndEventRemainsInSpool() + { + var spool = new EventSpool(_spoolDir, 10); + var sender = new BlockingUntilCanceledSender(); + var client = new MixpanelClient(); + client.InitializeForTest(spool, sender); + + client.Track("user-1", "Save", null); + + var stopwatch = Stopwatch.StartNew(); + var completed = Task.Run(() => client.ShutDown()).Wait(TimeSpan.FromSeconds(10)); + stopwatch.Stop(); + + Assert.IsTrue(completed, "ShutDown() did not return within the outer test timeout"); + Assert.Less(stopwatch.Elapsed, TimeSpan.FromSeconds(10), + "ShutDown() must be bounded even against a server that never responds"); + Assert.GreaterOrEqual(sender.CallCount, 1); + + // ShutDown() must have released the spool's cross-process lock -- reopening it and + // finding the event proves nothing was lost (it was rolled back, not flushed) and that + // the lock really was released. + using (var reopened = new EventSpool(_spoolDir, 10)) + { + Assert.AreEqual(1, reopened.ApproximateCount, + "the undelivered event must remain in the spool after a bounded shutdown"); + } + } + + [Test] + public void Flush_SenderBlocksUntilCanceled_ReturnsWithinBoundAndEventRemainsInSpool() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + var sender = new BlockingUntilCanceledSender(); + var client = new MixpanelClient(); + client.InitializeForTest(spool, sender); + + client.Track("user-1", "Save", null); + + var stopwatch = Stopwatch.StartNew(); + var completed = Task.Run(() => client.Flush()).Wait(TimeSpan.FromSeconds(10)); + stopwatch.Stop(); + + Assert.IsTrue(completed, "Flush() did not return within the outer test timeout"); + Assert.Less(stopwatch.Elapsed, TimeSpan.FromSeconds(10), + "Flush() must be bounded even against a server that never responds"); + Assert.AreEqual(1, spool.ApproximateCount, + "the undelivered event must remain in the spool after a bounded flush"); + } + } + + // ---- ASYNC PUBLIC SURFACE (FlushAsync/ShutDownAsync) ------------------------------------- + // + // The async counterparts must honor the same bounds as their sync wrappers. Awaited + // directly (no Task.Run + outer timeout) -- if the bound regressed, these would hang the + // awaiting test and fail via the test runner's timeout, same signal as the sync tests. + + [Test] + public async Task ShutDownAsync_SenderBlocksUntilCanceled_ReturnsWithinBoundAndEventRemainsInSpool() + { + var spool = new EventSpool(_spoolDir, 10); + var sender = new BlockingUntilCanceledSender(); + var client = new MixpanelClient(); + client.InitializeForTest(spool, sender); + + client.Track("user-1", "Save", null); + + var stopwatch = Stopwatch.StartNew(); + await client.ShutDownAsync(); + stopwatch.Stop(); + + Assert.Less(stopwatch.Elapsed, TimeSpan.FromSeconds(10), + "ShutDownAsync() must be bounded even against a server that never responds"); + Assert.GreaterOrEqual(sender.CallCount, 1); + + // ShutDownAsync() must have released the spool's cross-process lock -- reopening it and + // finding the event proves nothing was lost (it was rolled back, not flushed) and that + // the lock really was released. + using (var reopened = new EventSpool(_spoolDir, 10)) + { + Assert.AreEqual(1, reopened.ApproximateCount, + "the undelivered event must remain in the spool after a bounded shutdown"); + } + } + + [Test] + public async Task FlushAsync_SenderBlocksUntilCanceled_ReturnsWithinBoundAndEventRemainsInSpool() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + var sender = new BlockingUntilCanceledSender(); + var client = new MixpanelClient(); + client.InitializeForTest(spool, sender); + + client.Track("user-1", "Save", null); + + var stopwatch = Stopwatch.StartNew(); + await client.FlushAsync(); + stopwatch.Stop(); + + Assert.Less(stopwatch.Elapsed, TimeSpan.FromSeconds(10), + "FlushAsync() must be bounded even against a server that never responds"); + Assert.AreEqual(1, spool.ApproximateCount, + "the undelivered event must remain in the spool after a bounded flush"); + } + } + + // ShutDownAsync followed by Dispose (the natural shape for a host that awaits ShutDownAsync + // inside a `using` over the Analytics facade) must be a harmless no-op the second time. + [Test] + public async Task ShutDownAsync_ThenSyncShutDown_IsIdempotent() + { + var spool = new EventSpool(_spoolDir, 10); + var client = new MixpanelClient(); + client.InitializeForTest(spool, new AlwaysResultSender(SendResult.Delivered)); + + client.Track("user-1", "Save", null); + + await client.ShutDownAsync(); + Assert.DoesNotThrow(() => client.ShutDown()); + + // The spool's cross-process lock must be released (and stay released) -- reopening + // proves it. + using (var reopened = new EventSpool(_spoolDir, 10)) + { + Assert.AreEqual(0, reopened.ApproximateCount, + "the event was delivered during the first shutdown's bounded drain"); + } + } + + // ---- PER-EVENT SIZE CAP ------------------------------------------------------------------- + + // An event over the spool's per-event byte cap (production: Mixpanel's documented 1MB + // limit) is refused at Track time: never spooled, never handed to the sender, and counted + // as Failed -- NOT left to wedge the head of the queue as an eternally-retryable timeout. + [Test] + public async Task Track_EventOverSizeCap_IsDroppedCountedFailedAndNeverReachesSender() + { + using (var spool = new EventSpool(_spoolDir, 10, maxItemBytes: 500)) + { + var sender = new AlwaysResultSender(SendResult.Delivered); + var client = new MixpanelClient(); + client.InitializeForTest(spool, sender); + + client.Track("user-1", "Exception", + new JsonObject { { "Stack Trace", new string('x', 2000) } }); + + Assert.AreEqual(0, spool.ApproximateCount, "the oversized event must never be spooled"); + Assert.AreEqual(1, client.Statistics.Submitted); + Assert.AreEqual(1, client.Statistics.Failed); + + await client.DrainOnceAsync(); + Assert.AreEqual(0, sender.CallCount, "the oversized event must never reach the sender"); + + // A normal event on the same client still flows end to end. + client.Track("user-1", "Save", null); + await client.DrainOnceAsync(); + Assert.AreEqual(1, sender.CallCount); + Assert.AreEqual(1, client.Statistics.Succeeded); + } + } + + // A drain tick is byte-budgeted (bandwidth courtesy on slow/metered connections): three + // ~200KB events against the 256KB/tick budget take two ticks, not one. + [Test] + public async Task DrainOnce_EventsExceedingPerTickByteBudget_SpreadsAcrossTicks() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + var sender = new AlwaysResultSender(SendResult.Delivered); + var client = new MixpanelClient(); + client.InitializeForTest(spool, sender); + + for (var i = 0; i < 3; i++) + client.Track("user-1", "Big-" + i, + new JsonObject { { "Payload", new string('x', 200 * 1024) } }); + + await client.DrainOnceAsync(); + Assert.AreEqual(2, sender.CallCount, + "the 256KB/tick budget is crossed after the second ~200KB event"); + Assert.AreEqual(1, spool.ApproximateCount); + + await client.DrainOnceAsync(); + Assert.AreEqual(3, sender.CallCount); + Assert.AreEqual(0, spool.ApproximateCount); + } + } + + // ---- STATISTICS ------------------------------------------------------------------------- + + [Test] + public void Track_IncrementsSubmittedImmediately() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + var client = new MixpanelClient(); + client.InitializeForTest(spool, new AlwaysResultSender(SendResult.Delivered)); + + client.Track("user-1", "A", null); + client.Track("user-1", "B", null); + + Assert.AreEqual(2, client.Statistics.Submitted); + } + } + + // ---- POLLY -------------------------------------------------------------------------------- + // + // NOTE on which of the two documented approaches this uses (see offline-analytics.md, + // "Polly"): rather than driving Polly's real (non-zero) backoff delays through a + // FakeTimeProvider -- which is documented to hit the Polly #1932 + // SynchronizationContext-vs-ConfigureAwait gotcha -- this test injects a real retry pipeline + // (via MixpanelClient.BuildDefaultPipeline) configured with a ZERO base delay. That keeps + // the retry *strategy* itself real (ShouldHandle, MaxRetryAttempts, backoff wiring) while + // making the test deterministic and instantaneous, per the spec's documented fallback. + + [Test] + public async Task DrainOnce_WithRealRetryPipelineAndZeroDelay_TransientThenSuccess_DeliversWithinOneDrainOnceCall() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + var sender = new ScriptedSender(new Func[] + { + _ => SendResult.RetryableFailure, + _ => SendResult.Delivered + }); + var pipeline = MixpanelClient.BuildDefaultPipeline(TimeProvider.System, + maxRetryAttempts: 2, retryDelay: TimeSpan.Zero); + + var client = new MixpanelClient(); + client.InitializeForTest(spool, sender, pipeline: pipeline); + + client.Track("user-1", "Save", null); + await client.DrainOnceAsync(); + + Assert.AreEqual(0, spool.ApproximateCount, + "Polly should have retried in-process and delivered within this single DrainOnce call"); + Assert.AreEqual(2, sender.Sent.Count, + "the sender should have been called twice: once for the transient failure, once for the retry"); + Assert.AreEqual(1, client.Statistics.Succeeded); + } + } + + [Test] + public async Task DrainOnce_WithRealCircuitBreakerAndZeroDelay_SustainedFailures_TripsBreakerAndStopsCallingSender() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + var sender = new AlwaysResultSender(SendResult.RetryableFailure); + // MinimumThroughput=3 by default (see BuildDefaultPipeline): force enough failing + // attempts through quickly (small batch, zero delay) to trip the breaker, then verify + // a further drain of a fresh event does not even reach the sender. + // MaxRetryAttempts must be >= 1 (Polly validates this); 1 is close enough to "no + // retry" for this test's purposes -- what matters is that sender calls stop growing + // once the breaker trips, not the exact attempt count. + var pipeline = MixpanelClient.BuildDefaultPipeline(TimeProvider.System, + maxRetryAttempts: 1, retryDelay: TimeSpan.Zero); + + var client = new MixpanelClient(); + client.InitializeForTest(spool, sender, pipeline: pipeline, batchSize: 1); + + for (var i = 0; i < 8; i++) + { + client.Track("user-1", "Save-" + i, null); + await client.DrainOnceAsync(); + } + + var callsAfterWarmup = sender.CallCount; + + client.Track("user-1", "OneMore", null); + await client.DrainOnceAsync(); + + // Once the breaker is open, MixpanelClient must treat BrokenCircuitException as a + // RetryableFailure (leaving the event spooled) WITHOUT invoking the sender again. + Assert.AreEqual(callsAfterWarmup, sender.CallCount, + "once the circuit breaker is open, the sender must not be called again"); + Assert.Greater(spool.ApproximateCount, 0); + } + } + + [Test] + public async Task DrainOnce_WithRealCircuitBreakerAndFixedMinimumThroughput_OpensAfterOneFullyFailingTick() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + var sender = new AlwaysResultSender(SendResult.RetryableFailure); + // Regression test for the bug where MinimumThroughput=4 made the breaker inert: a + // fully failing DrainOnce tick produces exactly 3 outcomes (1 attempt + the default + // MaxRetryAttempts=2 retries), and successive ticks are 30s apart -- far outside the + // 10s sampling window -- so 4 outcomes could never land in the same window during a + // real outage. With MinimumThroughput fixed to 3, a single fully-failing tick alone + // must be enough to trip the breaker. + var pipeline = MixpanelClient.BuildDefaultPipeline(TimeProvider.System, retryDelay: TimeSpan.Zero); + + var client = new MixpanelClient(); + client.InitializeForTest(spool, sender, pipeline: pipeline, batchSize: 1); + + client.Track("user-1", "Save", null); + await client.DrainOnceAsync(); + + var callsAfterFirstTick = sender.CallCount; + Assert.GreaterOrEqual(callsAfterFirstTick, 3, + "one fully-failing tick should make at least 3 attempts (1 try + 2 retries)"); + + client.Track("user-1", "OneMore", null); + await client.DrainOnceAsync(); + + Assert.AreEqual(callsAfterFirstTick, sender.CallCount, + "the breaker should already be open after just one failing tick, so the sender must not be called again"); + Assert.Greater(spool.ApproximateCount, 0); + } + } + + // ---- CONTRACT: host parameter is rejected (Mixpanel does not support a host) ----------- + + [Test] + public void Initialize_WithNonEmptyHost_ThrowsArgumentException() + { + var client = new MixpanelClient(); + // Validated before any spool/timer is created, so no side effects and it propagates to the + // caller exactly as the original MixpanelClient did. + Assert.Throws(() => client.Initialize("secret", "https://example.com")); + } + + // ---- CONSENT PAUSE/RESUME (fix: re-enabling consent re-arms the flush loop) ------------- + + [Test] + public void PurgeQueuedEvents_ThenResumeSending_TogglesSendingPaused() + { + using (var spool = new EventSpool(_spoolDir, 10)) + { + var client = new MixpanelClient(); + client.InitializeForTest(spool, new AlwaysResultSender(SendResult.Delivered)); + + Assert.IsFalse(client.SendingPaused, "should start unpaused"); + + client.PurgeQueuedEvents(); + Assert.IsTrue(client.SendingPaused, "revoking consent (purge) must pause the flush loop"); + + client.ResumeSending(); + Assert.IsFalse(client.SendingPaused, "re-granting consent (resume) must un-pause the flush loop"); + } + } + } +} diff --git a/src/DesktopAnalyticsTests/MixpanelEventSenderTests.cs b/src/DesktopAnalyticsTests/MixpanelEventSenderTests.cs new file mode 100644 index 0000000..dccb9c3 --- /dev/null +++ b/src/DesktopAnalyticsTests/MixpanelEventSenderTests.cs @@ -0,0 +1,262 @@ +using System; +using System.Linq; +using System.Net.Http; +using System.Threading; +using System.Threading.Tasks; +using DesktopAnalytics; +using NUnit.Framework; +using WireMock.RequestBuilders; +using WireMock.ResponseBuilders; +using WireMock.Server; + +namespace DesktopAnalyticsTests +{ + // Layer 4 of the fidelity ladder in offline-analytics.md: a local HTTP stub (WireMock.Net) + // standing in for Mixpanel's real endpoint, exercising MixpanelEventSender's real HTTP + // posting/serialization and its response->SendResult classification. + [TestFixture] + public class MixpanelEventSenderTests + { + private WireMockServer _server; + + [SetUp] + public void SetUp() + { + _server = WireMockServer.Start(); + } + + [TearDown] + public void TearDown() + { + try + { + _server?.Stop(); + _server?.Dispose(); + } + catch (Exception e) + { + Console.WriteLine("MixpanelEventSenderTests.TearDown: " + e); + } + } + + private static AnalyticsEvent MakeEvent(string name = "Save") + { + return AnalyticsEvent.Create("user-1", name, null, "insert-id-1", + DateTimeOffset.UtcNow); + } + + [Test] + public async Task Send_200Response_ReturnsDelivered() + { + _server.Given(Request.Create().WithPath("/track").UsingPost()) + .RespondWith(Response.Create().WithStatusCode(200)); + + var sender = new MixpanelEventSender("secret", _server.Urls[0]); + + Assert.AreEqual(SendResult.Delivered, await sender.SendAsync(MakeEvent())); + } + + [Test] + public async Task Send_500Response_ReturnsRetryableFailure() + { + _server.Given(Request.Create().WithPath("/track").UsingPost()) + .RespondWith(Response.Create().WithStatusCode(500)); + + var sender = new MixpanelEventSender("secret", _server.Urls[0]); + + Assert.AreEqual(SendResult.RetryableFailure, await sender.SendAsync(MakeEvent())); + } + + [Test] + public async Task Send_429Response_ReturnsRetryableFailure() + { + _server.Given(Request.Create().WithPath("/track").UsingPost()) + .RespondWith(Response.Create().WithStatusCode(429)); + + var sender = new MixpanelEventSender("secret", _server.Urls[0]); + + Assert.AreEqual(SendResult.RetryableFailure, await sender.SendAsync(MakeEvent())); + } + + [Test] + public async Task Send_408Response_ReturnsRetryableFailure() + { + _server.Given(Request.Create().WithPath("/track").UsingPost()) + .RespondWith(Response.Create().WithStatusCode(408)); + + var sender = new MixpanelEventSender("secret", _server.Urls[0]); + + Assert.AreEqual(SendResult.RetryableFailure, await sender.SendAsync(MakeEvent())); + } + + [Test] + public async Task Send_400Response_ReturnsPoisonDrop() + { + _server.Given(Request.Create().WithPath("/track").UsingPost()) + .RespondWith(Response.Create().WithStatusCode(400)); + + var sender = new MixpanelEventSender("secret", _server.Urls[0]); + + Assert.AreEqual(SendResult.PoisonDrop, await sender.SendAsync(MakeEvent())); + } + + [Test] + public async Task Send_401Response_ReturnsPoisonDrop() + { + _server.Given(Request.Create().WithPath("/track").UsingPost()) + .RespondWith(Response.Create().WithStatusCode(401)); + + var sender = new MixpanelEventSender("secret", _server.Urls[0]); + + Assert.AreEqual(SendResult.PoisonDrop, await sender.SendAsync(MakeEvent())); + } + + [Test] + public async Task Send_ConnectionRefused_ReturnsRetryableFailure() + { + // Stop (and dispose) the server so the base URL it handed out is now unreachable -- + // simulates being offline without waiting for a real timeout (connection-refused is + // near-instant on loopback). + var deadUrl = _server.Urls[0]; + _server.Stop(); + _server.Dispose(); + _server = null; + + var sender = new MixpanelEventSender("secret", deadUrl, + new HttpClient { Timeout = TimeSpan.FromSeconds(5) }); + + Assert.AreEqual(SendResult.RetryableFailure, await sender.SendAsync(MakeEvent())); + } + + [Test] + public async Task Send_UnreachableHost_ReturnsRetryableFailure() + { + // A bad port on a live host is another common "offline" shape distinct from a + // stopped-server connection refusal. + var sender = new MixpanelEventSender("secret", "http://127.0.0.1:1", + new HttpClient { Timeout = TimeSpan.FromSeconds(5) }); + + Assert.AreEqual(SendResult.RetryableFailure, await sender.SendAsync(MakeEvent())); + } + + // ---- CAPTIVE PORTAL (redirect handling) -------------------------------------------------- + + [Test] + public async Task Send_302Response_ReturnsRetryableFailureAndDoesNotFollowRedirect() + { + // Simulates a captive portal intercepting the POST and redirecting to its login page. + _server.Given(Request.Create().WithPath("/track").UsingPost()) + .RespondWith(Response.Create() + .WithStatusCode(302) + .WithHeader("Location", "/portal-login")); + _server.Given(Request.Create().WithPath("/portal-login")) + .RespondWith(Response.Create().WithStatusCode(200)); + + var sender = new MixpanelEventSender("secret", _server.Urls[0]); + + Assert.AreEqual(SendResult.RetryableFailure, await sender.SendAsync(MakeEvent())); + + // The key assertion: AllowAutoRedirect must be disabled, so the portal's login page is + // never actually requested. If it were followed, this sender would see the portal's 200 + // and wrongly classify the event as Delivered even though Mixpanel never received it. + Assert.IsFalse(_server.LogEntries.Any(e => e.RequestMessage.Path == "/portal-login"), + "the redirect target must never have been requested (AllowAutoRedirect must be false)"); + } + + // ---- MIXPANEL'S "0" REJECTION SENTINEL ---------------------------------------------------- + + [Test] + public async Task Send_200ResponseWithBodyZero_ReturnsPoisonDrop() + { + // Mixpanel's /track returns HTTP 200 with a body of "0" for a payload it rejected. + _server.Given(Request.Create().WithPath("/track").UsingPost()) + .RespondWith(Response.Create().WithStatusCode(200).WithBody("0")); + + var sender = new MixpanelEventSender("secret", _server.Urls[0]); + + Assert.AreEqual(SendResult.PoisonDrop, await sender.SendAsync(MakeEvent())); + } + + [Test] + public async Task Send_200ResponseWithBodyOne_ReturnsDelivered() + { + _server.Given(Request.Create().WithPath("/track").UsingPost()) + .RespondWith(Response.Create().WithStatusCode(200).WithBody("1")); + + var sender = new MixpanelEventSender("secret", _server.Urls[0]); + + Assert.AreEqual(SendResult.Delivered, await sender.SendAsync(MakeEvent())); + } + + // ---- CANCELLATION (bounded drain deadline) ------------------------------------------------ + + [Test] + public void Send_PreCancelledToken_ReturnsRetryableFailureAndDoesNotThrow() + { + _server.Given(Request.Create().WithPath("/track").UsingPost()) + .RespondWith(Response.Create().WithStatusCode(200)); + + var sender = new MixpanelEventSender("secret", _server.Urls[0]); + + SendResult result = SendResult.Delivered; + using (var cts = new CancellationTokenSource()) + { + cts.Cancel(); + Assert.DoesNotThrowAsync(async () => result = await sender.SendAsync(MakeEvent(), cts.Token)); + } + + Assert.AreEqual(SendResult.RetryableFailure, result); + } + + [Test] + public void Send_ShortDeadlineAgainstDelayedResponse_ReturnsRetryableFailureAndDoesNotThrow() + { + // Simulates a "black hole" server: it eventually responds, but not before our deadline + // (BoundedDrain's per-attempt CancellationTokenSource, in production) fires. + _server.Given(Request.Create().WithPath("/track").UsingPost()) + .RespondWith(Response.Create().WithStatusCode(200).WithDelay(TimeSpan.FromSeconds(5))); + + var sender = new MixpanelEventSender("secret", _server.Urls[0], + new HttpClient { Timeout = TimeSpan.FromSeconds(30) }); + + SendResult result = SendResult.Delivered; + using (var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(200))) + { + Assert.DoesNotThrowAsync(async () => result = await sender.SendAsync(MakeEvent(), cts.Token)); + } + + Assert.AreEqual(SendResult.RetryableFailure, result); + } + + [Test] + public async Task Send_PostsFormEncodedBodyContainingTokenAndEventNameAndInsertId() + { + const string token = "my-token"; + + // The stub only matches (and returns 200) if the POST body is the expected + // "data=" shape and contains the token/event/insert-id we gave it -- + // this exercises MixpanelEventSender's real payload serialization end-to-end without + // coupling the test to Segment.Serialization's exact JSON formatting (key order, + // whitespace, etc). + _server + .Given(Request.Create() + .WithPath("/track") + .UsingPost() + .WithBody(body => + body != null && + body.StartsWith("data=") && + body.Contains(Uri.EscapeDataString("\"event\"")) && + body.Contains(Uri.EscapeDataString("\"Save PDF\"")) && + body.Contains(Uri.EscapeDataString("\"token\"")) && + body.Contains(Uri.EscapeDataString("\"" + token + "\"")) && + body.Contains(Uri.EscapeDataString("\"$insert_id\"")) && + body.Contains(Uri.EscapeDataString("\"insert-id-1\"")))) + .RespondWith(Response.Create().WithStatusCode(200)); + + var sender = new MixpanelEventSender(token, _server.Urls[0]); + + Assert.AreEqual(SendResult.Delivered, await sender.SendAsync(MakeEvent("Save PDF")), + "expected the WireMock stub -- which only matches the expected form body -- to have matched"); + } + } +} diff --git a/src/DesktopAnalyticsTests/PathScrubberTests.cs b/src/DesktopAnalyticsTests/PathScrubberTests.cs new file mode 100644 index 0000000..e047c23 --- /dev/null +++ b/src/DesktopAnalyticsTests/PathScrubberTests.cs @@ -0,0 +1,117 @@ +using DesktopAnalytics; +using NUnit.Framework; + +namespace DesktopAnalyticsTests +{ + [TestFixture] + public class PathScrubberTests + { + [Test] + public void Scrub_WindowsPath_ReplacesUserSegment() + { + const string input = @"C:\Users\jsmith\AppData\Local\Temp\file.txt"; + const string expected = @"%USER%\AppData\Local\Temp\file.txt"; + Assert.AreEqual(expected, PathScrubber.Scrub(input)); + } + + [Test] + public void Scrub_UnixPath_ReplacesUserSegment() + { + const string input = "/home/jsmith/.config/app/settings.json"; + const string expected = "%USER%/.config/app/settings.json"; + Assert.AreEqual(expected, PathScrubber.Scrub(input)); + } + + [Test] + public void Scrub_MultipleOccurrencesInOneString_ReplacesAll() + { + const string input = + "at Foo.Bar() in C:\\Users\\jsmith\\src\\Foo.cs:line 10\r\n" + + " at Baz.Qux() in C:\\Users\\otherUser\\src\\Baz.cs:line 20"; + const string expected = + "at Foo.Bar() in %USER%\\src\\Foo.cs:line 10\r\n" + + " at Baz.Qux() in %USER%\\src\\Baz.cs:line 20"; + Assert.AreEqual(expected, PathScrubber.Scrub(input)); + } + + [Test] + public void Scrub_MixedWindowsAndUnixInOneString_ReplacesBoth() + { + const string input = @"C:\Users\jsmith\a.txt and /home/jsmith/b.txt"; + const string expected = "%USER%\\a.txt and %USER%/b.txt"; + Assert.AreEqual(expected, PathScrubber.Scrub(input)); + } + + [Test] + public void Scrub_NoMatch_ReturnsInputUnchanged() + { + const string input = @"D:\Projects\MyApp\bin\Debug\App.exe"; + Assert.AreEqual(input, PathScrubber.Scrub(input)); + } + + [Test] + public void Scrub_PlainStringWithNoPaths_ReturnsInputUnchanged() + { + const string input = "Just a regular exception message, nothing to scrub here."; + Assert.AreEqual(input, PathScrubber.Scrub(input)); + } + + [Test] + public void Scrub_Null_ReturnsNull() + { + Assert.IsNull(PathScrubber.Scrub(null)); + } + + [Test] + public void Scrub_Empty_ReturnsEmpty() + { + Assert.AreEqual(string.Empty, PathScrubber.Scrub(string.Empty)); + } + + [Test] + public void Scrub_LowercaseDriveLetter_StillMatches() + { + const string input = @"c:\Users\jsmith\file.txt"; + const string expected = @"%USER%\file.txt"; + Assert.AreEqual(expected, PathScrubber.Scrub(input)); + } + + [Test] + public void Scrub_LowercaseUsersSegment_StillMatches() + { + const string input = @"C:\users\jsmith\file.txt"; + const string expected = @"%USER%\file.txt"; + Assert.AreEqual(expected, PathScrubber.Scrub(input)); + } + + [Test] + public void Scrub_MixedCaseDriveAndUsersSegment_StillMatches() + { + const string input = @"D:\UseRs\JSmith\file.txt"; + const string expected = @"%USER%\file.txt"; + Assert.AreEqual(expected, PathScrubber.Scrub(input)); + } + + [Test] + public void Scrub_UserNameWithSpacesAndDots_ReplacesWholeSegment() + { + const string input = @"C:\Users\John Q. Smith\Documents\file.txt"; + const string expected = @"%USER%\Documents\file.txt"; + Assert.AreEqual(expected, PathScrubber.Scrub(input)); + } + + [Test] + public void Scrub_HugeStackTraceWithManyPaths_ReplacesAllWithoutError() + { + var sb = new System.Text.StringBuilder(); + for (int i = 0; i < 500; i++) + { + sb.AppendLine($@" at Some.Method{i}() in C:\Users\jsmith\repo\File{i}.cs:line {i}"); + } + var result = PathScrubber.Scrub(sb.ToString()); + StringAssert.DoesNotContain("jsmith", result); + StringAssert.Contains("%USER%\\repo\\File0.cs", result); + StringAssert.Contains("%USER%\\repo\\File499.cs", result); + } + } +} diff --git a/src/SampleApp/Program.cs b/src/SampleApp/Program.cs index 99da2af..901255a 100644 --- a/src/SampleApp/Program.cs +++ b/src/SampleApp/Program.cs @@ -10,30 +10,60 @@ class Program { static int Main(string[] args) { - if (args.Length < 2 || args.Length > 3) + const string usage = "Usage: SampleApp " + + "[i[nitialTrackingState]=true|false] [c[onsentToggle]=true|false]"; + + if (args.Length < 2 || args.Length > 4) { - Console.WriteLine("Usage: SampleApp [i[nitialTrackingState]=true|false]"); + Console.WriteLine(usage); return 1; } if (!Enum.TryParse(args[1], true, out var clientType)) { - Console.WriteLine($"Usage: SampleApp {Environment.NewLine}Unrecognized client type: {args[1]}"); + Console.WriteLine($"{usage}{Environment.NewLine}Unrecognized client type: {args[1]}"); return 1; } var initialTracking = true; - - if (args.Length == 3) + // Exercises the AllowTracking off/on demo below by default (existing behavior). A caller + // driving this app as a durability-test harness across multiple process runs (rather than + // as a manual consent-flow demo) needs c=false: toggling AllowTracking calls + // PurgeQueuedEvents, which empties the ENTIRE on-disk spool immediately -- including any + // leftover events from a prior run -- so with the toggle left on, this app can never be + // used to prove "events spooled by a previous run survive and drain," only to demo consent. + var exerciseConsentToggle = true; + + for (var i = 2; i < args.Length; i++) { - var parts = args[2].Split('='); - if (parts.Length != 2 || - (!parts[0].Equals("i", StringComparison.OrdinalIgnoreCase) && - !parts[0].Equals("initialTrackingState", StringComparison.OrdinalIgnoreCase)) || - !bool.TryParse(parts[1], out initialTracking)) + var parts = args[i].Split('='); + if (parts.Length != 2) { - Console.WriteLine( - $"Unrecognized parameter: {args[2]}{Environment.NewLine}Usage: SampleApp [i[nitialTrackingState]=true|false]"); + Console.WriteLine($"Unrecognized parameter: {args[i]}{Environment.NewLine}{usage}"); + return 1; + } + + if (parts[0].Equals("i", StringComparison.OrdinalIgnoreCase) || + parts[0].Equals("initialTrackingState", StringComparison.OrdinalIgnoreCase)) + { + if (!bool.TryParse(parts[1], out initialTracking)) + { + Console.WriteLine($"Unrecognized parameter: {args[i]}{Environment.NewLine}{usage}"); + return 1; + } + } + else if (parts[0].Equals("c", StringComparison.OrdinalIgnoreCase) || + parts[0].Equals("consentToggle", StringComparison.OrdinalIgnoreCase)) + { + if (!bool.TryParse(parts[1], out exerciseConsentToggle)) + { + Console.WriteLine($"Unrecognized parameter: {args[i]}{Environment.NewLine}{usage}"); + return 1; + } + } + else + { + Console.WriteLine($"Unrecognized parameter: {args[i]}{Environment.NewLine}{usage}"); return 1; } } @@ -60,12 +90,16 @@ static int Main(string[] args) Debug.WriteLine("Sleeping for 20 seconds to give it all a chance to send an event in the background..."); Thread.Sleep(20000); - Analytics.AllowTracking = !Analytics.AllowTracking; - Analytics.Track("Should not be tracked"); - Debug.WriteLine("Sleeping for 2 seconds just for fun"); - Thread.Sleep(2000); + if (exerciseConsentToggle) + { + Analytics.AllowTracking = !Analytics.AllowTracking; + Analytics.Track("Should not be tracked"); + Debug.WriteLine("Sleeping for 2 seconds just for fun"); + Thread.Sleep(2000); + + Analytics.AllowTracking = !Analytics.AllowTracking; + } - Analytics.AllowTracking = !Analytics.AllowTracking; Analytics.SetApplicationProperty("TimeSinceLaunch", "25 seconds"); Analytics.Track("SomeEvent", new Dictionary {{"SomeValue", "42"}}); Console.WriteLine("Sleeping for another 20 seconds to give it all a chance to send an event in the background...");