Skip to content

Commit cab2c97

Browse files
louis4lilouis4li
committed
feat:update event
1 parent 512c077 commit cab2c97

1 file changed

Lines changed: 260 additions & 23 deletions

File tree

‎src/Aevatar.EventSourcing.Core/Storage/LogViewAdaptor.EventSourcing.cs‎

Lines changed: 260 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -264,7 +264,21 @@ private async Task ReadStateAsync(ViewStateSnapshot<TLogView> snapshot)
264264
if (_grainStorage != null)
265265
{
266266
var grainId = Services.GrainId.IsDefault ? ((IGrain)_host).GetGrainId() : Services.GrainId;
267-
await _grainStorage.ReadStateAsync(_grainTypeName, grainId, snapshot);
267+
268+
// Enhanced Orleans compatibility: Try framework format first, fall back to Orleans format
269+
try
270+
{
271+
Services.Log(LogLevel.Debug, "Attempting to read framework format snapshot for grain {GrainId}", grainId);
272+
await _grainStorage.ReadStateAsync(_grainTypeName, grainId, snapshot);
273+
Services.Log(LogLevel.Debug, "Successfully read framework format snapshot: RecordExists={RecordExists}", snapshot.RecordExists);
274+
}
275+
catch (Exception ex) when (IsOrleansFormatException(ex))
276+
{
277+
Services.Log(LogLevel.Information, "Framework format read failed with FormatException, attempting Orleans format conversion: {Error}", ex.Message);
278+
279+
// Try Orleans format reading when framework format fails
280+
await ReadOrleansFormatAndConvertAsync(grainId, snapshot, ex);
281+
}
268282
}
269283
else
270284
{
@@ -276,6 +290,135 @@ private async Task ReadStateAsync(ViewStateSnapshot<TLogView> snapshot)
276290
}
277291
}
278292

293+
/// <summary>
294+
/// Check if the exception indicates Orleans format incompatibility
295+
/// </summary>
296+
private static bool IsOrleansFormatException(Exception ex)
297+
{
298+
return ex is FormatException formatEx &&
299+
(formatEx.Message.Contains("Input string was not in a correct format") ||
300+
formatEx.Message.Contains("Expected an ASCII digit") ||
301+
formatEx.Message.Contains("Failure to parse"));
302+
}
303+
304+
/// <summary>
305+
/// Read Orleans format data and convert to framework format
306+
/// </summary>
307+
private async Task ReadOrleansFormatAndConvertAsync(GrainId grainId, ViewStateSnapshot<TLogView> snapshot, Exception originalException)
308+
{
309+
try
310+
{
311+
// Use Orleans LogStateWithMetaDataAndETag to read Orleans format directly
312+
var orleansLogState = new Orleans.EventSourcing.LogStorage.LogStateWithMetaDataAndETag<TLogEntry>();
313+
Services.Log(LogLevel.Information, "🔄 Reading Orleans LogStateWithMetaDataAndETag format for grain {GrainId}", grainId);
314+
315+
await _grainStorage.ReadStateAsync(_grainTypeName, grainId, orleansLogState);
316+
317+
Services.Log(LogLevel.Information, "Orleans format read result: RecordExists={RecordExists}, LogCount={LogCount}, GlobalVersion={GlobalVersion}",
318+
orleansLogState.RecordExists,
319+
orleansLogState.State?.Log?.Count ?? 0,
320+
orleansLogState.State?.GlobalVersion ?? 0);
321+
322+
if (orleansLogState.RecordExists && orleansLogState.State?.Log != null)
323+
{
324+
Services.Log(LogLevel.Information, "✅ Successfully read Orleans format data, converting to framework format");
325+
326+
// Convert Orleans data to framework snapshot format
327+
var convertedSnapshot = await ConvertOrleansToFrameworkSnapshot(orleansLogState);
328+
329+
// Copy converted data to output snapshot
330+
snapshot.RecordExists = true;
331+
snapshot.State = convertedSnapshot.State;
332+
snapshot.ETag = convertedSnapshot.ETag;
333+
334+
Services.Log(LogLevel.Information, "🎉 Orleans→Framework snapshot conversion completed: Version={Version}, WriteVector='{WriteVector}'",
335+
snapshot.State?.SnapshotVersion ?? 0, snapshot.State?.WriteVector ?? "empty");
336+
}
337+
else
338+
{
339+
Services.Log(LogLevel.Information, "No Orleans data found, initializing empty framework snapshot");
340+
341+
// Initialize empty framework snapshot
342+
snapshot.RecordExists = false;
343+
snapshot.State = new ViewStateSnapshotWithMetadata<TLogView>
344+
{
345+
Snapshot = new TLogView(),
346+
SnapshotVersion = 0,
347+
WriteVector = string.Empty
348+
};
349+
}
350+
}
351+
catch (Exception orleansEx)
352+
{
353+
Services.Log(LogLevel.Warning, "Orleans format reading also failed: {Error}, falling back to empty snapshot", orleansEx.Message);
354+
Services.Log(LogLevel.Debug, "Original framework error: {FrameworkError}", originalException.Message);
355+
Services.Log(LogLevel.Debug, "Orleans reading error: {OrleansError}", orleansEx.ToString());
356+
357+
// Initialize empty snapshot when both formats fail
358+
snapshot.RecordExists = false;
359+
snapshot.State = new ViewStateSnapshotWithMetadata<TLogView>
360+
{
361+
Snapshot = new TLogView(),
362+
SnapshotVersion = 0,
363+
WriteVector = string.Empty
364+
};
365+
}
366+
}
367+
368+
/// <summary>
369+
/// Convert Orleans LogStateWithMetaDataAndETag to Framework ViewStateSnapshot
370+
/// </summary>
371+
private async Task<ViewStateSnapshot<TLogView>> ConvertOrleansToFrameworkSnapshot(Orleans.EventSourcing.LogStorage.LogStateWithMetaDataAndETag<TLogEntry> orleansLogState)
372+
{
373+
var frameworkSnapshot = new ViewStateSnapshot<TLogView>();
374+
375+
// Replay Orleans events to build current state
376+
var currentView = new TLogView();
377+
var version = 0;
378+
379+
if (orleansLogState.State?.Log != null)
380+
{
381+
foreach (var logEntry in orleansLogState.State.Log)
382+
{
383+
try
384+
{
385+
_host.UpdateView(currentView, logEntry);
386+
version++;
387+
Services.Log(LogLevel.Debug, "Applied Orleans event {EventIndex}: {EventType}", version, logEntry.GetType().Name);
388+
}
389+
catch (Exception ex)
390+
{
391+
Services.CaughtUserCodeException("UpdateView", nameof(ConvertOrleansToFrameworkSnapshot), ex);
392+
Services.Log(LogLevel.Warning, "Failed to apply Orleans event {EventIndex}: {Error}", version + 1, ex.Message);
393+
}
394+
}
395+
}
396+
397+
// Convert Orleans WriteVector to framework format
398+
var orleansWriteVector = orleansLogState.State?.WriteVector ?? string.Empty;
399+
var frameworkWriteVector = ConvertOrleansWriteVectorToFrameworkFormat(orleansWriteVector);
400+
401+
// Create framework snapshot
402+
frameworkSnapshot.RecordExists = true;
403+
frameworkSnapshot.ETag = orleansLogState.ETag ?? string.Empty;
404+
frameworkSnapshot.State = new ViewStateSnapshotWithMetadata<TLogView>
405+
{
406+
Snapshot = DeepCopy(currentView),
407+
SnapshotVersion = version,
408+
WriteVector = frameworkWriteVector
409+
};
410+
411+
// Update internal state
412+
_confirmedView = DeepCopy(currentView);
413+
_confirmedVersion = version;
414+
_globalVersion = Math.Max(version, orleansLogState.State?.GlobalVersion ?? 0);
415+
416+
Services.Log(LogLevel.Information, "Orleans→Framework conversion: {EventCount} events → Version {Version}, GlobalVersion {GlobalVersion}",
417+
orleansLogState.State?.Log?.Count ?? 0, version, _globalVersion);
418+
419+
return frameworkSnapshot;
420+
}
421+
279422
private async Task WriteStateAsync()
280423
{
281424
if (_grainStorage != null)
@@ -319,74 +462,168 @@ private async Task ProcessFrameworkDataAsync(GrainId grainId)
319462
}
320463

321464
/// <summary>
322-
/// Try to convert Orleans memory event structure to MongoDB snapshot
465+
/// Enhanced Orleans compatibility: Convert Orleans Memory EventSourcing format to Framework MongoDB format
466+
/// This method is called when reading framework format fails, to provide seamless Orleans→Framework compatibility
323467
/// </summary>
324468
private async Task TryConvertOrleansLogStorageAsync(GrainId grainId)
325469
{
326-
Services.Log(LogLevel.Information, "TryConvertOrleansLogStorageAsync called for grain {GrainId}", grainId);
470+
Services.Log(LogLevel.Information, "🔄 Enhanced Orleans compatibility: Attempting Orleans→Framework conversion for grain {GrainId}", grainId);
327471

328472
if (_grainStorage == null)
329473
{
330-
Services.Log(LogLevel.Information, "No grain storage available, using initial state");
474+
Services.Log(LogLevel.Information, "No grain storage available for Orleans conversion, using initial state");
331475
return;
332476
}
333477

334478
try
335479
{
336-
// Use Orleans LogStateWithMetaDataAndETag class directly
480+
// Use Orleans LogStateWithMetaDataAndETag class directly to read Orleans format
337481
var orleansLogState = new Orleans.EventSourcing.LogStorage.LogStateWithMetaDataAndETag<TLogEntry>();
338-
Services.Log(LogLevel.Information, "Attempting to read Orleans LogStorage for grain {GrainId}", grainId);
482+
Services.Log(LogLevel.Information, "Reading Orleans LogStorage format for grain {GrainId}", grainId);
339483

340484
await _grainStorage.ReadStateAsync(_grainTypeName, grainId, orleansLogState);
341485

342-
Services.Log(LogLevel.Information, "Orleans LogStorage read result: RecordExists={RecordExists}, LogCount={LogCount}",
343-
orleansLogState.RecordExists, orleansLogState.State?.Log?.Count ?? 0);
486+
Services.Log(LogLevel.Information, "Orleans LogStorage read result: RecordExists={RecordExists}, LogCount={LogCount}, GlobalVersion={GlobalVersion}",
487+
orleansLogState.RecordExists,
488+
orleansLogState.State?.Log?.Count ?? 0,
489+
orleansLogState.State?.GlobalVersion ?? 0);
344490

345-
if (orleansLogState.RecordExists && orleansLogState.State.Log.Count > 0)
491+
if (orleansLogState.RecordExists && orleansLogState.State?.Log != null)
346492
{
347-
Services.Log(LogLevel.Information, "Found Orleans LogStorage with {Count} events, converting to snapshot",
348-
orleansLogState.State.Log.Count);
493+
var eventCount = orleansLogState.State.Log.Count;
494+
var globalVersion = orleansLogState.State.GlobalVersion;
349495

350-
// Rebuild state from Orleans event log
496+
Services.Log(LogLevel.Information, "✅ Found Orleans data: {EventCount} events, GlobalVersion={GlobalVersion}, converting to Framework format",
497+
eventCount, globalVersion);
498+
499+
// Initialize fresh state for rebuilding
351500
_confirmedView = new TLogView();
352501
_confirmedVersion = 0;
353502

503+
// Replay Orleans events to rebuild state
354504
foreach (var logEntry in orleansLogState.State.Log)
355505
{
356506
try
357507
{
358508
_host.UpdateView(_confirmedView, logEntry);
359509
_confirmedVersion++;
510+
Services.Log(LogLevel.Debug, "Applied Orleans event {EventIndex}: {EventType}",
511+
_confirmedVersion, logEntry.GetType().Name);
360512
}
361513
catch (Exception ex)
362514
{
363515
Services.CaughtUserCodeException("UpdateView", nameof(TryConvertOrleansLogStorageAsync), ex);
516+
Services.Log(LogLevel.Warning, "Failed to apply Orleans event {EventIndex}: {Error}",
517+
_confirmedVersion + 1, ex.Message);
364518
}
365519
}
366520

367-
_globalVersion = _confirmedVersion;
521+
// Set global version from Orleans data
522+
_globalVersion = Math.Max(_confirmedVersion, globalVersion);
368523

369-
// Create framework snapshot
524+
// Create framework snapshot with Orleans data
370525
_globalSnapshot.State.Snapshot = DeepCopy(_confirmedView);
371526
_globalSnapshot.State.SnapshotVersion = _confirmedVersion;
372-
// Reset WriteVector to empty string to avoid format compatibility issues
373-
// between Orleans WriteVector format and framework WriteVector format
374-
_globalSnapshot.State.WriteVector = string.Empty;
375527

376-
Services.Log(LogLevel.Information, "Converted Orleans LogStorage to snapshot: {EventCount} events, version {Version}",
377-
orleansLogState.State.Log.Count, _confirmedVersion);
528+
// Convert Orleans WriteVector format to Framework format
529+
var orleansWriteVector = orleansLogState.State.WriteVector ?? string.Empty;
530+
_globalSnapshot.State.WriteVector = ConvertOrleansWriteVectorToFrameworkFormat(orleansWriteVector);
531+
532+
Services.Log(LogLevel.Information, "🎉 Orleans→Framework conversion completed successfully: {EventCount} events replayed, version {Version}, WriteVector '{WriteVector}'",
533+
eventCount, _confirmedVersion, _globalSnapshot.State.WriteVector);
534+
535+
// Save the converted data in framework format for future use
536+
try
537+
{
538+
await WriteStateAsync();
539+
Services.Log(LogLevel.Information, "✅ Converted Orleans data saved in Framework format for future access");
540+
}
541+
catch (Exception ex)
542+
{
543+
Services.Log(LogLevel.Warning, "Failed to save converted Orleans data: {Error}", ex.Message);
544+
}
378545
}
379546
else
380547
{
381-
Services.Log(LogLevel.Information, "No Orleans LogStorage found (RecordExists={RecordExists}), using initial state",
548+
Services.Log(LogLevel.Information, "No Orleans events found (RecordExists={RecordExists}), initializing with empty state",
382549
orleansLogState.RecordExists);
550+
551+
// Initialize empty state if no Orleans data
552+
_confirmedView = new TLogView();
553+
_confirmedVersion = 0;
554+
_globalVersion = 0;
555+
_globalSnapshot.State.Snapshot = DeepCopy(_confirmedView);
556+
_globalSnapshot.State.SnapshotVersion = 0;
557+
_globalSnapshot.State.WriteVector = string.Empty;
383558
}
384559
}
560+
catch (FormatException ex) when (ex.Message.Contains("Input string was not in a correct format"))
561+
{
562+
// This is the specific error we're trying to fix
563+
Services.Log(LogLevel.Error, "❌ Orleans WriteVector FormatException detected: {Error}", ex.Message);
564+
Services.Log(LogLevel.Information, "Initializing with empty state due to Orleans format incompatibility");
565+
566+
// Initialize empty state when Orleans format parsing fails
567+
_confirmedView = new TLogView();
568+
_confirmedVersion = 0;
569+
_globalVersion = 0;
570+
_globalSnapshot.State.Snapshot = DeepCopy(_confirmedView);
571+
_globalSnapshot.State.SnapshotVersion = 0;
572+
_globalSnapshot.State.WriteVector = string.Empty;
573+
}
574+
catch (Exception ex)
575+
{
576+
Services.Log(LogLevel.Warning, "Orleans LogStorage reading failed: {Error}", ex.Message);
577+
Services.Log(LogLevel.Debug, "Orleans compatibility exception details: {ExceptionDetails}", ex.ToString());
578+
579+
// Initialize empty state if Orleans reading fails for any other reason
580+
_confirmedView = new TLogView();
581+
_confirmedVersion = 0;
582+
_globalVersion = 0;
583+
_globalSnapshot.State.Snapshot = DeepCopy(_confirmedView);
584+
_globalSnapshot.State.SnapshotVersion = 0;
585+
_globalSnapshot.State.WriteVector = string.Empty;
586+
}
587+
}
588+
589+
/// <summary>
590+
/// Convert Orleans WriteVector format to Framework compatible format
591+
/// Orleans format: ",replica1,replica2" -> Framework format: "replica1;replica2"
592+
/// </summary>
593+
private string ConvertOrleansWriteVectorToFrameworkFormat(string orleansWriteVector)
594+
{
595+
if (string.IsNullOrEmpty(orleansWriteVector))
596+
return string.Empty;
597+
598+
try
599+
{
600+
// Orleans WriteVector typically starts with comma: ",replica1,replica2"
601+
if (orleansWriteVector.StartsWith(","))
602+
{
603+
var converted = orleansWriteVector.TrimStart(',').Replace(",", ";");
604+
Services.Log(LogLevel.Debug, "Converted Orleans WriteVector '{Orleans}' to Framework format '{Framework}'",
605+
orleansWriteVector, converted);
606+
return converted;
607+
}
608+
609+
// If it doesn't start with comma, check if it needs conversion
610+
if (orleansWriteVector.Contains(","))
611+
{
612+
var converted = orleansWriteVector.Replace(",", ";");
613+
Services.Log(LogLevel.Debug, "Converted Orleans WriteVector '{Orleans}' to Framework format '{Framework}'",
614+
orleansWriteVector, converted);
615+
return converted;
616+
}
617+
618+
// Already in compatible format
619+
Services.Log(LogLevel.Debug, "Orleans WriteVector '{WriteVector}' is already in compatible format", orleansWriteVector);
620+
return orleansWriteVector;
621+
}
385622
catch (Exception ex)
386623
{
387-
Services.Log(LogLevel.Warning, "Failed to read Orleans LogStorage: {Exception}", ex.Message);
388-
Services.Log(LogLevel.Debug, "Orleans LogStorage exception details: {ExceptionDetails}", ex.ToString());
389-
// If reading fails, continue with initial state
624+
Services.Log(LogLevel.Warning, "Failed to convert Orleans WriteVector '{WriteVector}': {Error}, using empty",
625+
orleansWriteVector, ex.Message);
626+
return string.Empty;
390627
}
391628
}
392629
}

0 commit comments

Comments
 (0)